我使用梁 python 庫設(shè)計了一個梁/數(shù)據(jù)流管道。管道大致執(zhí)行以下操作：ParDo：從 API 收集 JSON 數(shù)據(jù)ParDo：轉(zhuǎn)換 JSON 數(shù)據(jù)I/O：將轉(zhuǎn)換后的數(shù)據(jù)寫入 BigQuery 表一般來說，代碼會做它應(yīng)該做的事情。但是，當(dāng)從 API 收集一個大數(shù)據(jù)集（大約 500.000 個 JSON 文件）時，bigquery 插入作業(yè)在使用 DataflowRunner（它與在我的計算機(jī)）。使用較小的數(shù)據(jù)集時，一切正常。數(shù)據(jù)流日志如下：2019-04-22 (00:41:29) Executing BigQuery import job "dataflow_job_14675275193414385105". You can check its status with the...Executing BigQuery import job "dataflow_job_14675275193414385105". You can check its status with the bq tool: "bq show -j --project_id=X dataflow_job_14675275193414385105". 2019-04-22 (00:41:29) Workflow failed. Causes: S01:Create Dummy Element/Read+Call API+Transform JSON+Write to Bigquery /Wr...Workflow failed. Causes: S01:Create Dummy Element/Read+Call API+Transform JSON+Write to Bigquery /WriteToBigQuery/NativeWrite failed., A work item was attempted 4 times without success. Each time the worker eventually lost contact with the service. The work item was attempted on: beamapp-X-04212005-04211305-sf4k-harness-lqjg,beamapp-X-04212005-04211305-sf4k-harness-lgg2,beamapp-X-04212005-04211305-sf4k-harness-qn55,beamapp-X-04212005-04211305-sf4k-harness-hcsn按照建議使用 bq cli 工具來獲取有關(guān) BQ 加載作業(yè)的更多信息不起作用。找不到工作（我懷疑它是否已經(jīng)創(chuàng)建，因?yàn)榧磿r失敗）。我想我遇到了某種配額/bq 限制，甚至是內(nèi)存不足問題（請參閱：https ://beam.apache.org/documentation/io/built-in/google-bigquery/ ）限制 BigQueryIO 目前有以下限制。您無法使用 >您的管道的其他步驟對 BigQuery 寫入的完成進(jìn)行排序。如果您使用的是 Beam SDK for Python，如果您編寫了一個非常大的數(shù)據(jù)集，您可能會遇到導(dǎo)入大小配額>問題。作為一種解決方法，您可以對數(shù)據(jù)集進(jìn)行分區(qū)（例如，使用 Beam 的分區(qū)轉(zhuǎn)換）并寫入多個 BigQuery 表。Beam SDK for Java 沒有這個>限制，因?yàn)樗鼤槟鷦澐謹(jǐn)?shù)據(jù)集。對于如何縮小此問題的根本原因，我將不勝感激。我還想嘗試分區(qū) Fn，但沒有找到任何 python 源代碼示例如何將分區(qū) pcollection 寫入 BigQuery 表。
2 回答
寶慕林4294392
TA貢獻(xiàn)2021條經(jīng)驗(yàn) 獲得超8個贊
可能有助于調(diào)試的一件事是查看 Stackdriver 日志。
如果您在 Google控制臺中打開 Dataflow 作業(yè)并單擊LOGS圖形面板右上角的 ，則應(yīng)該會打開底部的日志面板。LOGS 面板的右上角有一個指向 Stackdriver 的鏈接。這將為您提供有關(guān)您的工人/洗牌/等的大量日志信息。對于這個特定的工作。
其中有很多內(nèi)容，并且很難過濾掉相關(guān)的內(nèi)容，但希望您能夠找到比A work item was attempted 4 times without success. 例如，每個工作人員偶爾會記錄它正在使用的內(nèi)存量，這可以與每個工作人員擁有的內(nèi)存量（基于機(jī)器類型）進(jìn)行比較，以查看它們是否確實(shí)內(nèi)存不足，或者您的錯誤是否正在發(fā)生別處。
祝你好運(yùn)！
富國滬深
TA貢獻(xiàn)1790條經(jīng)驗(yàn) 獲得超9個贊
據(jù)我所知，在 Cloud Dataflow 和 Apache Beam 的 Python SDK 中沒有可用的選項來診斷 OOM（Java SDK 可以）。我建議您在Cloud Dataflow 問題跟蹤器中打開功能請求，以獲取有關(guān)此類問題的更多詳細(xì)信息。
除了檢查 Dataflow 作業(yè)日志文件之外，我建議您使用Stackdriver Monitoring 工具監(jiān)控您的管道，該工具提供每個作業(yè)的資源使用情況（作為總內(nèi)存使用時間）。
關(guān)于 Python SDK 中的 Partition 函數(shù)使用，以下代碼（基于 Apache Beam文檔中提供的示例）將數(shù)據(jù)拆分為 3 個 BigQuery 加載作業(yè)：
def partition_fn(input_data, num_partitions):
return int(get_percentile(lines) * num_partitions / 100)
partition = input_data | beam.Partition(partition_fn, 3)
for x in range(3):
partition[x] | 'WritePartition %s' % x >> beam.io.WriteToBigQuery(
table_spec,
schema=table_schema,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
添加回答
舉報
0/150
提交
取消