Dataflow – 将 JSON 文件添加到 BigQuery

huangapple
go评论113阅读模式
英文:

Dataflow - add JSON file to BigQuery

问题

我正在使用GCP Dataflow进行一些POC，并向BigQuery添加一些JSON对象。

  1. import apache_beam as beam
  2. import apache_beam.io.gcp.bigquery as b_query
  4. p1 = beam.Pipeline()
  5. trips_schema = 'trip_id:INTEGER,vendor_id:INTEGER,trip_distance:FLOAT,fare_amount:STRING,store_and_fwd_flag:STRING'
  7. freq = (
  8.         p1
  9.         | beam.Create([{"vendor_id":33,"trip_id": 1000474,"trip_distance": 2.3999996185302734,"fare_amount": 42.13,"store_and_fwd_flag": "Y"}])
  10.         | beam.Map(print)
  11.         | 'Write Record to BigQuery' >> b_query.WriteToBigQuery(table='trips2', dataset='my_poc',
  12.                                                                 custom_gcs_temp_location='gs://XXXX-stage-xxxx/temp',
  13.                                                                 schema=trips_schema, project='xxxxxxxx-xxx-2',
  14.                                                                 write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
  15.                                                                 create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
  16. )
  17. p1.run()

现在，当我运行此代码时，我收到以下错误：

  1. RuntimeError: BigQuery job beam_bq_job_LOAD_AUTOMATIC_JOB_NAME_LOAD_STEP_957_6db0f0222c18bf6ef55dfb301cf9b7b2_2e2519daf47e48888bd08fc7661da2e6 failed. Error Result: <ErrorProto
  2.  location: 'gs://xxxx-stage-xxx/temp/bq_load/4b49a738d79b47ba8b188f04d10bf8f0/xxxx-dev-x.my_poc.trips2/258cab64-9d98-4fbc-8c16-087bdd0ea93c'
  3.  message: 'Error while reading data, error message: JSON table encountered too many errors, giving up. Rows: 1; errors: 1. Please look into the errors[] collection for more details. File: gs://xxxx-stage-xxx/temp/bq_load/4b49a738d79b47ba8b188f04d10bf8f0/xxxxx-dev-x.my_poc.trips2/258cab64-9d98-4fbc-8c16-087bdd0ea93c'
  4.  reason: 'invalid'> [while running 'Write Record to BigQuery/BigQueryBatchFileLoads/TriggerLoadJobsWithoutTempTables/ParDo(TriggerLoadJobs)']
  6. 在引用错误的阶段桶中的文件为空。
  8. 请帮助。
  10. <details>
  11. <summary>英文:</summary>
  13. I&#39;m doing some POC with GCP Dataflow and add some JSON object to BigQuery. 
  15.     import apache_beam as beam
  16.     import apache_beam.io.gcp.bigquery as b_query
  17.     
  18.     p1 = beam.Pipeline()
  19.     trips_schema = &#39;trip_id:INTEGER,vendor_id:INTEGER,trip_distance:FLOAT,fare_amount:STRING,store_and_fwd_flag:STRING&#39;
  20.     
  21.     freq = (
  22.             p1
  23.             | beam.Create([&quot;{\&quot;vendor_id\&quot;:33,\&quot;trip_id\&quot;: 1000474,\&quot;trip_distance\&quot;: 2.3999996185302734,\&quot;fare_amount\&quot;: 42.13,\&quot;store_and_fwd_flag\&quot;: \&quot;Y\&quot;}&quot;])
  24.             | beam.Map(print)
  25.             | &#39;Write Record to BigQuery&#39; &gt;&gt; b_query.WriteToBigQuery(table=&#39;trips2&#39;, dataset=&#39;my_poc&#39;,
  26.                                                                     custom_gcs_temp_location=&#39;gs://XXXX-stage&#39;
  27.                                                                                              &#39;-xxxx/temp&#39;,
  28.                                                                     schema=trips_schema, project=&#39;xxxxxxxx-xxx-2&#39;,
  29.                                                                     write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
  30.                                                                     create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, )
  31.     )
  32.     p1.run()
  34. Now when I&#39;m running this code, I&#39;m getting the following error:
  36.     RuntimeError: BigQuery job beam_bq_job_LOAD_AUTOMATIC_JOB_NAME_LOAD_STEP_957_6db0f0222c18bf6ef55dfb301cf9b7b2_2e2519daf47e48888bd08fc7661da2e6 failed. Error Result: &lt;ErrorProto
  37.      location: &#39;gs://xxxx-stage-xxx/temp/bq_load/4b49a738d79b47ba8b188f04d10bf8f0/xxxx-dev-x.my_poc.trips2/258cab64-9d98-4fbc-8c16-087bdd0ea93c&#39;
  38.      message: &#39;Error while reading data, error message: JSON table encountered too many errors, giving up. Rows: 1; errors: 1. Please look into the errors[] collection for more details. File: gs://xxxx-stage-xxx/temp/bq_load/4b49a738d79b47ba8b188f04d10bf8f0/xxxxx-dev-x.my_poc.trips2/258cab64-9d98-4fbc-8c16-087bdd0ea93c&#39;
  39.      reason: &#39;invalid&#39;&gt; [while running &#39;Write Record to BigQuery/BigQueryBatchFileLoads/TriggerLoadJobsWithoutTempTables/ParDo(TriggerLoadJobs)&#39;]
  41. File in the stage bucket which referred in error contains null. 
  43. Please help.
  45. </details>
  48. # 答案1
  49. **得分**: 2
  51. 如果您想使用`Beam`和`Dataflow`将`Json`文件插入到`BigQuery`表中，您必须以`NEWLINE_DELIMITED_JSON`格式读取`Json`文件，示例：
  53. 您的代码示例，以模拟输入为例：
  55. ```python
  56. import apache_beam as beam
  57. import apache_beam.io.gcp.bigquery as b_query
  59. def map_to_dict(input: str) -> Dict:
  60.     return json.loads(input)
  62. p1 = beam.Pipeline()
  63. trips_schema = 'trip_id:INTEGER,vendor_id:INTEGER,trip_distance:FLOAT,fare_amount:STRING,store_and_fwd_flag:STRING'
  65. inputs: List[str] = [
  66.     '{"vendor_id":33,"trip_id": 1000474,"trip_distance": 2.3999996185302734,"fare_amount": 42.13,"store_and_fwd_flag": "Y"}',
  67.     '{"vendor_id":34,"trip_id": 1000475,"trip_distance": 2.3999996185302734,"fare_amount": 42.13,"store_and_fwd_flag": "Y"}'
  68. ]
  70. freq = (
  71.     p1
  72.     | beam.Create(inputs)
  73.     | beam.Map(print)
  74.     | beam.Map(map_to_dict)
  75.     | 'Write Record to BigQuery' >> b_query.WriteToBigQuery(table='trips2', dataset='my_poc',
  76.                                                             custom_gcs_temp_location='gs://XXXX-stage-xxxx/temp',
  77.                                                             schema=trips_schema, project='xxxxxxxx-xxx-2',
  78.                                                             write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
  79.                                                             create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
  80. )
  81. p1.run()

从GCS中的Json文件示例：

input.json：

  1. {"vendor_id":33,"trip_id": 1000474,"trip_distance": 2.3999996185302734,"fare_amount": 42.13,"store_and_fwd_flag": "Y"}
  2. {"vendor_id":34,"trip_id": 1000475,"trip_distance": 2.3999996185302734,"fare_amount": 42.13,"store_and_fwd_flag": "Y"}
  1. import apache_beam as beam
  2. import apache_beam.io.gcp.bigquery as b_query
  4. def map_to_dict(input: str) -> Dict:
  5.     return json.loads(input)
  7. p1 = beam.Pipeline()
  8. trips_schema = 'trip_id:INTEGER,vendor_id:INTEGER,trip_distance:FLOAT,fare_amount:STRING,store_and_fwd_flag:STRING'
  10. freq = (
  11.     p1
  12.     | ReadFromText("gs://your-bucket/object/input.json")
  13.     | beam.Map(print)
  14.     | beam.Map(map_to_dict)
  15.     | 'Write Record to BigQuery' >> b_query.WriteToBigQuery(table='trips2', dataset='my_poc',
  16.                                                             custom_gcs_temp_location='gs://XXXX-stage-xxxx/temp',
  17.                                                             schema=trips_schema, project='xxxxxxxx-xxx-2',
  18.                                                             write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
  19.                                                             create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
  20. )
  21. p1.run()

需要遵守以下条件：

  • 在将PCollection保存到BigQuery之前，需要将字符串转换为字典。
  • Json对象应与BigQuery表的架构完全匹配。您可以在WriteToBigQuery输出连接器中使用此参数来忽略额外的字段：ignore_unknown_columns
英文:

If you want to insert a Json file to a BigQuery table with Beam and Dataflow, you have to read the Json file with NEWLINE_DELIMITED_JSON format, example :

Your code with a mock as input :

  1. import apache_beam as beam
  2. import apache_beam.io.gcp.bigquery as b_query
  4. def map_to_dict(input: str) -&gt; Dict:
  5.     return json.loads(input)
  7. p1 = beam.Pipeline()
  8. trips_schema = &#39;trip_id:INTEGER,vendor_id:INTEGER,trip_distance:FLOAT,fare_amount:STRING,store_and_fwd_flag:STRING&#39;
  10. inputs: List[str] = [
  11.             &quot;{\&quot;vendor_id\&quot;:33,\&quot;trip_id\&quot;: 1000474,\&quot;trip_distance\&quot;: 2.3999996185302734,&quot;&quot;\&quot;fare_amount\&quot;: 42.13,\&quot;store_and_fwd_flag\&quot;: \&quot;Y\&quot;}&quot;,
  12.             &quot;{\&quot;vendor_id\&quot;:34,\&quot;trip_id\&quot;: 1000475,\&quot;trip_distance\&quot;: 2.3999996185302734,&quot;&quot;\&quot;fare_amount\&quot;: 42.13,\&quot;store_and_fwd_flag\&quot;: \&quot;Y\&quot;}&quot;
  13.         ]
  15. freq = (
  16.         p1
  17.         | beam.Create(inputs)
  18.         | beam.Map(print)
  19.         | beam.Map(map_to_dict)
  20.         | &#39;Write Record to BigQuery&#39; &gt;&gt; b_query.WriteToBigQuery(table=&#39;trips2&#39;, dataset=&#39;my_poc&#39;,
  21.                                                                 custom_gcs_temp_location=&#39;gs://XXXX-stage&#39;
  22.                                                                                          &#39;-xxxx/temp&#39;,
  23.                                                                 schema=trips_schema, project=&#39;xxxxxxxx-xxx-2&#39;,
  24.                                                                 write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
  25.                                                                 create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, )
  26. )
  27. p1.run()

Example with a Json file from GCS :

input.json :

  1. {&quot;vendor_id&quot;:33,&quot;trip_id&quot;: 1000474,&quot;trip_distance&quot;: 2.3999996185302734,&quot;fare_amount&quot;: 42.13,&quot;store_and_fwd_flag&quot;: &quot;Y&quot;}
  2. {&quot;vendor_id&quot;:34,&quot;trip_id&quot;: 1000475,&quot;trip_distance&quot;: 2.3999996185302734,&quot;fare_amount&quot;: 42.13,&quot;store_and_fwd_flag&quot;: &quot;Y&quot;}
  1. import apache_beam as beam
  2. import apache_beam.io.gcp.bigquery as b_query
  4. def map_to_dict(input: str) -&gt; Dict:
  5.     return json.loads(input)
  7. p1 = beam.Pipeline()
  8. trips_schema = &#39;trip_id:INTEGER,vendor_id:INTEGER,trip_distance:FLOAT,fare_amount:STRING,store_and_fwd_flag:STRING&#39;
  10. freq = (
  11.         p1
  12.         | ReadFromText(&quot;gs://your-bucket/object/input.json&quot;)
  13.         | beam.Map(print)
  14.         | beam.Map(map_to_dict)
  15.         | &#39;Write Record to BigQuery&#39; &gt;&gt; b_query.WriteToBigQuery(table=&#39;trips2&#39;, dataset=&#39;my_poc&#39;,
  16.                                                                 custom_gcs_temp_location=&#39;gs://XXXX-stage&#39;
  17.                                                                                          &#39;-xxxx/temp&#39;,
  18.                                                                 schema=trips_schema, project=&#39;xxxxxxxx-xxx-2&#39;,
  19.                                                                 write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
  20.                                                                 create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, )
  21. )
  22. p1.run()

The following conditions need to be respected :

  • You need to transform your string to Dict before to save the PCollection to BigQuery
  • The Json object should match exactly the schema of the BigQuery table. You can ignore extra fields with this parameter in the WriteToBigQuery output connector : ignore_unknown_columns

huangapple
  • 本文由 发表于 2023年6月5日 18:43:06
  • 转载请务必保留本文链接：https://go.coder-hub.com/76405650.html
  • apache-beam
  • google-cloud-dataflow
  • google-cloud-platform
  • python
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定