google-cloud-dataflow

1熱度

1回答

我們有什麼能夠提供DoFn的完成狀態，就像函數完成它的執行一樣嗎？我們是否可以生成任何觸發器或任何可以公正地完成DoFn函數內部所有步驟的公平想法？任何幫助將不勝感激。

0熱度

1回答

有什麼簡單的方法可以將GroupBy的輸出重定向到基於組密鑰的多個輸出文件？ Bin.apply(GroupByKey.<String, KV<Long,Iterable<TableRow>>>create()) .apply(ParDo.named("Print Bins").of(...) .apply(TextIO.Write.to(*Output file based on key*)

0熱度

1回答

對錶進行分區

Bigquery目前只允許按日期進行分區。 Lets supose我有一個與inserted_timestamp字段的10億錶行。讓我們說這個領域有1年前的日期。將現有數據移動到新的分區表的正確方法是什麼？編輯我看到有Java的一個優雅的解決方案與版本< 2.0 Sharding BigQuery output tables還闡述了在BigQuery partitioning with Be

0熱度

1回答

Google DataFlow無法在存儲區中寫入

我無法運行示例wordcount數據流。一個錯誤出現： (venv) [email protected]:$ python -m apache_beam.examples.wordcount \ --project "$PROJECT" --runner DataflowRunner \ --staging_location $BUCKET/staging \

0熱度

1回答

如何避免在Luigi內部使用Google-Cloud-Dataflow導入錯誤

我有多個相互依賴的進程。我正在使用Luigi來管理這些依賴關係。由於有多個流程，我已經對每個流程進行了包裝。 I.e進程的所有相關文件都位於具有init文件的文件夾中。我正在使用一個Luigi任務來查看它的依賴關係，並執行那個任務需要完成的進程。其中一項任務是數據流作業。當我將其稱爲數據流的工作是提出了一個導入錯誤，導入錯誤：沒有模塊名爲TaskBQ2DS.TaskBQ2DS (38c7b56

0熱度

1回答

測試PubsubIO讀取本地模擬器

我能夠設置TestPipeline和PubsubIO與Pubsub模擬器的工作： options = TestPipeline.testingPipelineOptions().as(PubsubOptions.class); options.setPubsubRootUrl(pubsubUrl); 但問題是，如何約束PubsubIO.readMessages()時，單元測試，實際上

0熱度

1回答

谷歌雲數據流Python - maxNumWorkers

我想增加用Apache Beam的python SDK構建的數據流管道的工作人員數量，並且我發現文檔建議設置--maxNumWorkers=標誌將足以增加超出最大數量的工作人員默認值是15.但是，當我將這個標誌添加到管道選項時，它似乎不起作用。我回頭看了文檔here中記錄的執行參數選項，並注意到maxNumWorkers未在Python「列出其他雲管道選項」中列出，但出現在Java SDK中 -

1熱度

1回答

在執行BigQueryIO.write（）時獲取/設置BigQuery作業ID

是否可以設置BigQuery作業ID或在批處理管道運行時獲取它。我知道使用BigQuery API是可能的，但是如果我使用Apache Beam中的BigQueryIO，可以嗎？在寫入BigQuery之後，我需要發送一個確認，確認加載完成。

1熱度

2回答

谷歌雲數據流從字典中寫入CSV

我有一個值的字典，我想使用Python SDK將它作爲有效的.CSV文件寫入GCS。我可以將字典寫成換行分隔的文本文件，但我似乎無法找到將字典轉換爲有效的.CSV的示例。任何人都可以建議在數據流管道中生成csv的最佳方法嗎？這回答了這個question地址從CSV文件讀取，但並沒有真正解決寫入CSV文件。我認識到CSV文件只是包含規則的文本文件，但我仍然在努力將數據字典轉換爲可以使用WriteTo

0熱度

1回答

阿帕奇梁 - 無法推斷在DOFN編碼器與多輸出標籤

我嘗試使用Apache束執行管線，但試圖把一些輸出標籤時，我得到一個錯誤： import com.google.cloud.Tuple; import com.google.gson.Gson; import com.google.gson.reflect.TypeToken; import org.apache.beam.sdk.Pipeline; import org.apache.b