【发布时间】：2020-02-21 04:13:58
【问题描述】：
我正在使用 Apache Beam 创建一个数据管道，以将 XML 文件作为来自谷歌存储桶 (GCS) 的输入并将其转换为 JSON 文件。我正在尝试使用 python 的“xmltodict”库首先将 XML 转换为 python dict，然后使用 python json.dumps() 函数将其转换为 json 格式。我为梁管道的每个步骤创建了单独的 beam.DoFn 类。
我在一个小文件（小于 1MB 大小）上测试了管道，它工作正常。代码在 directrunner（不到 1 分钟）和数据流运行器（5-6 分钟，包括数据流作业的启动和停止）上运行。但是，当我将数据流运行器与更大的文件（例如（~150 MB））一起使用时，管道会继续运行近 1 小时而没有任何进展。我不是要弄清楚什么是错的。
我认为将整个文件作为一个字符串的输入将是一个问题，如果我能以某种更好的方式从 GCS 读取 XML 文件以便将每个记录解析为单个记录，将解决这个问题。我将不胜感激任何有助于改善这一点的帮助。
以下是示例代码：
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions
from apache_beam.options.pipeline_options import StandardOptions
class ReadGCSfile(beam.DoFn):
def process(self,element):
from apache_beam.io.gcp import gcsio
gcs = gcsio.GcsIO()
yield(gcs.open(element).read())
# this class converts hyphens(-) in the XML text into (_) so that column name convention is inline with #the BigQuery conventions
class clean_xml(beam.DoFn):
def process(self,element):
if re.search('(<.[^\>]*)-', element) == None:
return[element]
else:
for i in range(len(re.findall('(<.[^\>]*)-', element))):
y = re.search('(<.[^\>]*)-', element)
splitpoint1 = y.span()[0]
splitpoint2 = y.span()[1]
element = element[:splitpoint1] + element[splitpoint1:splitpoint2].replace('-','_') + element[splitpoint2:]
return[(element)]
class createdict(beam.DoFn):
def process(self,element):
import xmltodict
order_data = xmltodict.parse(element)
unnested_data = order_data['root1']['root2']
return[(unnested_data)]
class converttojson(beam.DoFn):
def process(self,element):
import json
import re
for order in element:
order_j = json.dumps(order)
yield(order_j)
def run(argv = None):
options = PipelineOptions()
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = 'project123'
google_cloud_options.job_name = 'job123'
google_cloud_options.staging_location = 'gs://bucket123/staging'
google_cloud_options.temp_location = 'gs://bucket123/temp'
google_cloud_options.machine_type = 'n1-standard-8'
options.view_as(StandardOptions).runner = 'DataflowRunner'
p = beam.Pipeline(options=options)
input_file_path = 'gs://' + input_bucket +'/'+input_file
(p
| 'Create' >> beam.Create([input_file_path])
| 'GetXML' >> beam.ParDo(ReadGCSfile())
| 'Clean_XML' >> beam.ParDo(clean_xml())
| 'CreateDict' >> beam.ParDo(createdict())
| 'Convert2JSON' >> beam.ParDo(converttojson())
| 'write' >> beam.io.WriteToText('gs://bk/output',file_name_suffix='.json',num_shards =1,shard_name_template='')
)
p.run()
我可以对此进行哪些改进，以使其更有效地处理更大的文件。管道现在似乎停留在 GetXML 和 Clean_XML 阶段。如何一次遍历一条记录的 XML 文件？
下面是一个示例数据文件：
<?xml version="1.0" encoding="UTF-8"?>
<root1 xmlns="http://www.example.com">
<root2 ID-no="000000">
<date>2022-09-23T06:58:24.000Z</date>
<created-by>storefront</created-by>
<original-order-no>000000</original-order-no>
<currency>USD</currency>
<invoice-no>11111111</invoice-no>
<customer>
<customer-name>abcccccc</customer-name>
<customer-email>abccccc@gmail.com</customer-email>
<billing-address>
<address1>20 xyz</address1>
<city>mars</city>
<postal-code>123456</postal-code>
<state-code>hhjbjh</state-code>
<country-code>nm mn</country-code>
</billing-address>
</customer>
<status>
<order-status>NEW</order-status>
<shipping-status>NOT_SHIPPED</shipping-status>
<confirmation-status>CONFIRMED</confirmation-status>
<payment-status>NOT_PAID</payment-status>
</status>
</root2>
</root1>
【问题讨论】：
-
如果是单个大文件。我怀疑您在这里没有任何并行性：只有一个线程/工作者处理数据，因此速度很慢。在这种情况下，工人更有可能受到限制，从而使整个管道花费更长的时间。
-
这个有什么办法吗？
-
先问这个问题，有没有办法把这个单个文件拆分成多个，然后并行处理？如果答案是否定的，那么我认为没有一种简单的解决方法。
标签： xml python-2.7 google-cloud-platform google-cloud-dataflow apache-beam