如何通过自定义Query实现BigQuery UDF将数据迁移至MemoryStore？

一、整体迁移步骤规划 #

• 定义get_column_from_table(keys)：从BigQuery根据主键获取对应行数据
• 定义get_field_value(field, value)：处理字段值（含数组拆分逻辑），生成适合MemoryStore的格式
• 定义post()：将处理后的数据写入MemoryStore（执行hset操作）

二、各步骤具体实现建议 #

1. get_column_from_table(keys)函数实现 #

BigQuery原生UDF无法直接在函数内部执行跨表查询，推荐两种实现方式：

同表内提取主键对应字段（SQL UDF）

如果是从当前查询的表中根据主键提取字段，可直接用SQL UDF：

CREATE OR REPLACE FUNCTION `your-project.your-dataset.get_column_from_table`(key_col STRING, input_row ANY TYPE)
RETURNS STRING
LANGUAGE js AS """
  // input_row为当前表的行数据，key_col是主键列名
  return input_row[key_col];
""";

跨表查询主键数据（远程函数+Cloud Function）

如果需要从其他表查询主键对应数据，用BigQuery远程函数关联Cloud Function实现：

const {BigQuery} = require('@google-cloud/bigquery');
const bigquery = new BigQuery();

exports.getColumnFromTable = async (req, res) => {
  const {keys} = req.body;
  // 构建查询语句，根据主键集合查询目标表
  const query = `SELECT * FROM \`your-project.your-dataset.target-table\` WHERE id IN UNNEST(@keys)`;
  const options = {
    query: query,
    params: {keys: keys},
  };
  const [rows] = await bigquery.query(options);
  res.status(200).send(rows);
};

在BigQuery中创建远程函数关联该Cloud Function后，即可在SQL中直接调用。

2. get_field_value(field, value)函数优化（适配数组拆分） #

针对数组字段拆分需求，优化你提供的代码：

function get_field_value(field, element) {
  // 先处理数组类型的字段值，拆分为换行分隔的字符串
  for (const key in element) {
    if (Array.isArray(element[key])) {
      element[key] = element[key].join('\n');
    }
  }

  const keys = Object.keys(element);
  const uniq_keys = [...new Set(keys.map(x => x.split(':')[0]))];
  const result = {};
  
  for (const uniq of uniq_keys) {
    let value = null;
    for (const key of keys) {
      if (key.split(':')[0] === uniq) {
        value = value ? `${value}\n${element[key]}` : element[key];
      }
    }
    result[uniq] = value;
  }
  return result[field] || "";
}

3. post()函数实现（Cloud Function操作MemoryStore） #

通过Redis客户端执行hset操作写入MemoryStore：

const redis = require('redis');
const client = redis.createClient({
  host: 'your-memorystore-host',
  port: 6379,
});

exports.postToMemoryStore = async (req, res) => {
  const {key, fieldValues} = req.body;
  // fieldValues为键值对对象，格式：{field1: "value1", field2: "value2"}
  await client.hSet(key, fieldValues);
  res.status(200).send({status: 'success'});
};

三、Cloud Function灵活查询实现 #

实现支持仅输入主键获取全量数据（忽略Null）或指定列的功能：

const {BigQuery} = require('@google-cloud/bigquery');
const bigquery = new BigQuery();
const redis = require('redis');
const client = redis.createClient({host: 'your-memorystore-host', port: 6379});

exports.migrateData = async (req, res) => {
  const {key, field} = req.body;
  let selectClause;

  if (!field) {
    // 动态获取表所有字段，用IFNULL过滤Null值
    const [table] = await bigquery.dataset('your-dataset').table('your-table').get();
    const schemaFields = table.metadata.schema.fields.map(f => f.name);
    selectClause = schemaFields.map(col => `IFNULL(${col}, '') AS ${col}`).join(', ');
  } else {
    // 指定字段，同样过滤Null值
    selectClause = `IFNULL(${field}, '') AS ${field}`;
  }

  // 根据主键查询数据
  const query = `SELECT ${selectClause} FROM \`your-project.your-dataset.your-table\` WHERE id = @key`;
  const options = {query: query, params: {key: key}};
  const [rows] = await bigquery.query(options);

  if (rows.length === 0) {
    return res.status(404).send({message: 'No data found'});
  }

  const row = rows[0];
  const fieldValues = {};

  if (field) {
    fieldValues[field] = get_field_value(field, row);
  } else {
    for (const col in row) {
      fieldValues[col] = get_field_value(col, row);
    }
  }

  // 写入MemoryStore
  await client.hSet(key, fieldValues);
  res.status(200).send({status: 'success', data: fieldValues});
};

// 复用字段处理函数
function get_field_value(field, element) {
  for (const key in element) {
    if (Array.isArray(element[key])) {
      element[key] = element[key].join('\n');
    }
  }

  const keys = Object.keys(element);
  const uniq_keys = [...new Set(keys.map(x => x.split(':')[0]))];
  const result = {};
  
  for (const uniq of uniq_keys) {
    let value = null;
    for (const key of keys) {
      if (key.split(':')[0] === uniq) {
        value = value ? `${value}\n${element[key]}` : element[key];
      }
    }
    result[uniq] = value;
  }
  return result[field] || "";
}

请求示例 #

• 获取主键对应全量数据：POST /migrateData，请求体：{"key": "your-primary-key"}
• 获取主键对应指定列：POST /migrateData，请求体：{"key": "your-primary-key", "field": "target-column"}

内容的提问来源于stack exchange，提问作者George Alexandru Postolache