

現在、GCPのCloud Dataflowにて動かすためのパイプラインを  
Apache beamにてPythonを用いて作成しています。

パイプラインの中の処理として、2つの異なるBigQueryテーブルから読み込んだデータを基に、  
SQLのUPDATE文を利用して片方のテーブルを更新するということを実装したいと考えています。

そこで、SQLTransformにてPcollectionに対してSQL文を実行できるという記載を  
見つけたため、実際にSELECT文などを発行して試したところ、問題なく実行できました。

その後、簡単なUPDATE文も試してみたところ、こちらは以下のようなエラーとなりました。

[code]

    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])
    
[/code]

SQLTransformにてUPDATEを利用する際の文法自体が間違っていたり、  
BigQueryから読み込んだ後に必要な前処理などがありましたらご教授お願いします。

コード全文は以下となります。  
ReadFromBigQueryを利用してデータを読み込んだ後、Row型にマッピングを行ってSQLTransformを利用できる状態にしたのち、  
UPDATE文を試しています。

[code]

    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()
    
[/code]


CC BY-SA 4.0
2023-09-21T04:17:15.087
96295
2023-09-21T04:17:15.087
59857
question
python sql google-cloud
Apache BeamのSQLTransformにてUPDATE文が実行できない。