現在、GCPのCloud Dataflowにて動かすためのパイプラインを
Apache beamにてPythonを用いて作成しています。
パイプラインの中の処理として、2つの異なるBigQueryテーブルから読み込んだデータを基に、
SQLのUPDATE文を利用して片方のテーブルを更新するということを実装したいと考えています。
そこで、SQLTransformにてPcollectionに対してSQL文を実行できるという記載を
見つけたため、実際にSELECT文などを発行して試したところ、問題なく実行できました。
その後、簡単なUPDATE文も試してみたところ、こちらは以下のようなエラーとなりました。
RuntimeError: org.apache.beam.sdk.extensions.sql.impl.SqlConversionException: Unable to convert query
UPDATE
UpdateTable
SET
CreatedByName = 'test'
~中略~
at org.apache.beam.sdk.extensions.sql.impl.CalciteQueryPlanner.convertToBeamRel(CalciteQueryPlanner.java:214)
Caused by: org.apache.beam.vendor.calcite.v1_28_0.org.apache.calcite.plan.RelOptPlanner$CannotPlanException: There are not enough rules to produce a node with desired properties: convention=BEAM_LOGICAL.
Missing conversion is LogicalTableModify[convention: NONE -> BEAM_LOGICAL]
There is 1 empty subset: rel#12:RelSubset#2.BEAM_LOGICAL, the relevant part of the original plan is as follows
10:LogicalTableModify(table=[[beam, UpdateTable]], operation=[UPDATE], updateColumnList=[[CreatedByName]], sourceExpressionList=[['test']], flattened=[true])
SQLTransformにてUPDATEを利用する際の文法自体が間違っていたり、
BigQueryから読み込んだ後に必要な前処理などがありましたらご教授お願いします。
コード全文は以下となります。
ReadFromBigQueryを利用してデータを読み込んだ後、Row型にマッピングを行ってSQLTransformを利用できる状態にしたのち、
UPDATE文を試しています。
import apache_beam as beam
from apache_beam.io.gcp.internal.clients import bigquery
from apache_beam.transforms.sql import SqlTransform
from apache_beam.options.pipeline_options import PipelineOptions
def run():
import apache_beam as beam
options = PipelineOptions(
project="[projectId]",
temp_location="[gs://~]",
)
# データをReadするQueryを定義
ReadUpdateTableQuery="""
SELECT
*
FROM
[myproject.mydataset.updatetable]
"""
ReadUpdateSourceQuery="""
SELECT
*
FROM
[myproject.mydataset.updatesource]
"""
# Update対象のデータをreadし、SQLTransform可能な形に変換する。
ReadUpdateTable = (p | 'Read UpdateTable' >> beam.io.ReadFromBigQuery(query=ReadUpdateTableQuery,
method=beam.io.ReadFromBigQuery.Method.DIRECT_READ
)
| 'Modify UpdateTable' >> beam.Map(lambda elem: beam.Row(
name=str(elem['name']),
age=int(elem['age']),
address=str(elem['address']),
CreatedByName=str(elem['CreatedByName']),
)
)
)
# Updateに利用するためのデータをreadし、SQLTransform可能な形に変換する。
ReadUpdateSource = (p | 'Read UpdateSource' >> beam.io.ReadFromBigQuery(query=ReadUpdateSourceQuery,
method=beam.io.ReadFromBigQuery.Method.DIRECT_READ
)
| 'Modify UpdateSource' >> beam.Map(lambda elem: beam.Row(
name=str(elem['name']),
age=int(elem['age']),
CreatedByName=str(elem['CreatedByName'])
)
)
)
# SQLテストのためのSELECT文(問題なく動作)
# SQLQuery="""
# SELECT
# *
# FROM
# UpdateTable
# UPDATE文のテストSQL文(エラーが発生)
SQLQuery="""
UPDATE
UpdateTable
SET
CreatedByName = 'test'
WHERE true
"""
# 実際に行いたい処理
# SQLQuery="""
# UPDATE
# UpdateTable AS t
# SET
# t.age = (
# SELECT s.age
# FROM UpdateSource AS s
# WHERE t.name = s.name
# ),
# t.CreatedByName = (
# SELECT s.CreatedByName
# FROM UpdateSource AS s
# WHERE t.name = s.name
# )
# WHERE EXISTS (
# SELECT 1
# FROM UpdateSource AS s
# WHERE t.name = s.name)
# """
# Readした2つのPcollectionを利用してSQL文を実行する。
sql = ({"UpdateTable": ReadUpdateTable, "UpdateSource": ReadUpdateSource}
| SqlTransform(SQLQuery)
| beam.Map(print)
)
p.run()
if __name__ == '__main__':
run()