在 BigQuery 中运行 python 函数答案

【问题标题】：Running a python function in BigQuery在 BigQuery 中运行 python 函数
【发布时间】：2019-08-23 01:32:44
【问题描述】：

是否可以在 BigQuery 中运行 python 函数？

根据this blog post from Felipe，似乎 C 可以编译为 WebAssembly 并在 BQ 中运行。

当然，Python 可以使用 cython 或其他一些工具编译为 C 或 C++（或者甚至可以转译为 javascript）。那么我的问题是，有没有人有在 BigQuery 中执行 python 函数的经验。如果是这样，您使用的流程是什么？

这里可能的选项是：

将python“转换”成javascript运行。
将python编译成c或cpp并编译成wasm

这是一个可以使用的示例输入：

(1) 来源

id         product 
1          box     
2          bottle

(2) Python 函数使用

def double_id(row):
    return row['id'] * 2

def product_code(row):
    # B3
    return row['product'].upper()[0] + str(len(row['product']))

(3) 预期输出

id     product      double_id       product_code
1      box          2               B3
2      bottle       4               B6

我不只是想用 javascript 重写上面的代码（这可能是最简单的方法），但我正在寻找一种更通用的解决方案，如果存在的话——在哪里我可以使用 python（标准库）函数并在 BigQuery 查询中使用它。

【问题讨论】：

我没有这样做的经验。但是，根据当前的 BigQuery 功能，可能可以运行编译后的 javascript-python 函数或从基于 javascript 的 Python 解释器运行 python 脚本。这是link 和link。
您可以使用 Python API 查询 BigQuery，然后应用您喜欢的任何自定义代码操作：cloud.google.com/bigquery/docs/reference/libraries - 这是您要查找的内容吗？我还使用 Cloud Functions 来执行 BigQuery ETL 作业。
我很好奇 - python 相对于 JS 的额外好处是什么？
@RobertLacok 这是一个设计/UI 决定——从技术上讲，是的，JS 会更好。
根据您最终要完成的工作范围，有一个 BigQuery Storage API integration with pandas 允许您将查询结果下载到 DataFrame。这可能会增加一个您希望避免但最终让您能够运行 python 函数的额外步骤。

标签： python google-bigquery webassembly transpiler

【解决方案1】：

Python 3 Apache Beam + BigQuery 以下是从 BigQuery 读取和写入 BigQuery 的关键 Beam 代码：

with beam.Pipeline(RUNNER, options = opts) as p:
    (p 
      | 'read_bq' >> beam.io.Read(beam.io.BigQuerySource(query=query, use_standard_sql=True))
      | 'compute_fit' >> beam.FlatMap(compute_fit)
      | 'write_bq' >> beam.io.gcp.bigquery.WriteToBigQuery(
          'ch05eu.station_stats', schema='station_id:string,ag:FLOAT64,bg:FLOAT64,cg:FLOAT64')
    )

本质上，我们正在对 BigQuery 表运行查询，运行 Python 方法 compute_fit，并将输出写入 BigQuery 表。这是我的 compute_fit 方法。如您所见，这只是简单的 Python 代码：

def compute_fit(row):
  from scipy import stats
  import numpy as np
  durations = row['duration_array']
  ag, bg, cg = stats.gamma.fit(durations)
  if np.isfinite(ag) and np.isfinite(bg) and np.isfinite(cg):
      result = {}
      result['station_id'] = str(row['start_station_id'])
      result['ag'] = ag
      result['bg'] = bg
      result['cg'] = cg
      yield result

确保在 requirements.txt 中指定您需要在 Dataflow 工作器上安装的 Python 包：

%%writefile requirements.txt
numpy
scipy

享受吧！有关更多信息，您可以参考此文档How to run Python code on your BigQuery table

【讨论】：