GCP 数据流 UDF 导入文件答案

【问题标题】：GCP Dataflow UDF import filesGCP 数据流 UDF 导入文件
【发布时间】：2022-08-02 14:45:57
【问题描述】：

我正在使用 PubSub to Elastic Dataflow 模板，我需要做一些扩充，比如 XML 解析。

我有两个问题：

我可以将我的代码拆分为多个文件并使用 UDF 导入它们吗？如何？
我可以安装外部库吗？我只阅读了支持纯 js 文件，但随后阅读了这篇文章，我不确定是否适用于 Dataflow UDF 或如何实现它：https://medium.com/analytics-vidhya/using-npm-library-in-google-bigquery-udf-8aef01b868f4
任何指导将不胜感激。

谢谢

你能澄清一下你在这里谈论的是什么类型的UDF吗？您是否只是想在写入 Elastic Search 之前对模板处理的数据应用一些中间函数？
嗨@chamikara，我需要解析 xml 内容和 grok，我正在考虑为此导入库。尝试使用 webpack 构建文件，但没有成功。
要导入新库并添加新的 Beam DoFn（用于额外处理），您必须修改 Dataflow 模板或手动运行管道。任何其他库都可以通过在作业提交时将相关 jar 添加到 CLASSPATH 来简单地导入。默认情况下，CLASSPATH 中的所有 jar 都将由 Dataflow 暂存（如果您使用的是 Beam Python，请参阅beam.apache.org/documentation/sdks/python-pipeline-dependencies）
我正在考虑导入构建单个 js 文件的 javascript 库。
Beam 目前不提供直接使用 Javascript 库的方法。如果有关联的 Python/Java/Go 库，您也许可以使用它。请参阅 Apache Beam 网站了解更多详情 - beam.apache.org/get-started/beam-overview

标签： javascript google-cloud-dataflow user-defined-functions

【解决方案1】：

我快到了。我没有正确构建单个 js 文件。

src/index.js

var convert = require("xml-js");

module.exports.data_process = function (inJson) {
  var xml = "<XML CONTENT>";
  var result1 = convert.xml2json(xml, { compact: true, spaces: 4 });
  var parsed = JSON.parse(result1);
  var some_field = parsed.some.field
  var obj = JSON.parse(inJson);
  obj.some_field = some_field;
  return JSON.stringify(obj);
};

webpack.config.js

const path = require("path");

module.exports = {
  entry: "./src/index.ts",
  mode: "production",
  output: {
    filename: "main.js",
    path: path.resolve(__dirname, "dist"),
    library: "xmllib",
  },
};

构建 webpack

npx webpack --config webpack.config.js

然后添加到main.js的末尾

function process(d) {
    c = xmllib.data_process(d);
    return c;
}

现在您可以使用“process”作为您的数据流 udf 函数名称。

基于本文的解决方案：https://blog.salrashid.dev/articles/2020/bq-udf-protobuf/

【讨论】：