google-cloud-dataflow

0熱

1答えて

Google Cloud DataflowがGoogle Storageの親ディレクトリを見つけることができません

Streamflowモードでデータフローパイプラインを実行すると、java.io.IOExceptionはGoogle Storageの親ディレクトリを見つけることができません。 gsutil lsを使用してファイルを表示でき、パイプラインがバッチモードで実行されているときにこれが動作するため、これはアクセス権の問題ではないと思います。何が原因でしょうか？ [WARNING] java.lan

0熱

1答えて

Dataflowパイプラインの外部Python依存関係

python依存関係をGoogleクラウドデータフローパイプラインに読み込むことはできますか？私はgensim's phrase modelerを使用して、データを1行ずつ読み込んで、一般的なフレーズ/バイグラム（2つの単語が頻繁に隣に現れる）を自動的に検出します。したがって、パイプラインの最初の実行は、各フレーズをこのフレーズモデラーに渡すことになります。パイプラインを2度目に通過すると、同じフ

3熱

1答えて

Java Beam SDKを使用してデータフロージョブのステップの詳細を取得する方法は？

データフロージョブにJava Beam SDKを使用しています。com.google.api.services.dataflow.model.Jobクラスは、特定のジョブの詳細を示します。しかし、それは、そのような要素を追加しましたとしてデータフローのステップの情報を取得するために任意の方法/プロパティを提供していない、推定サイズなどの下は、私は仕事のを取得するために使用しているコードです。情報

0熱

1答えて

Dataflow GroupBy - >キーに基づく複数の出力

GroupByの出力をGroupキーに基づいて複数の出力ファイルにリダイレクトする簡単な方法はありますか？ Bin.apply(GroupByKey.<String, KV<Long,Iterable<TableRow>>>create()) .apply(ParDo.named("Print Bins").of(...) .apply(TextIO.Write.to(*Output file

0熱

1答えて

テーブルのパーティション化

Bigqueryでは、現時点では日付のみでパーティションを作成できます。私はフィールドと10億テーブルの行を持っています。このフィールドの有効期限は1年前です。既存のデータを新しいパーティションテーブルに移動する正しい方法は何ですか？は編集私は、データをウィンドウイングテーブル名（またはパーティションの接尾辞）をパラメータ化することである< 2.0 Sharding BigQuery ou

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データフローのインポートエラーを回避する方法

私はお互いに依存する複数のプロセスを持っています。私はこれらの依存関係を管理するためにLuigiを使用しています。複数のプロセスがあるので、私はそれぞれのプロセスのパッケージを作りました。つまり、プロセスのすべての関連ファイルは、initファイルを持つフォルダにあります。私はその依存関係を調べて実行するLuigiタスクを1つ使用しています。このタスクは実行するために完了する必要があります。これら

0熱

1答えて

テストPubsubIOは地元のエミュレータで読み取る

私は Pubsubエミュレータで動作するように設定 TestPipelineと PubsubIOすることができた： options = TestPipeline.testingPipelineOptions().as(PubsubOptions.class); options.setPubsubRootUrl(pubsubUrl); が、問題は、ユニットテストは、実際には有限のテス

0熱

1答えて

Google Cloud Dataflow Python --maxNumWorkers

Apache BeamのPython SDKを使用して構築されたデータフローパイプラインの作業者数を増やそうとしています。文書番号：--maxNumWorkers=フラグを設定すると、ただし、このフラグをパイプラインオプションに追加すると、動作していないように見えます。私はhereと記載されている実行パラメータオプションを振り返り、maxNumWorkersがPythonの "他のクラウドパイプラ

1熱

2答えて

Google Cloud Dataflow辞書からCSVに書き込む

Python SDKを使用して有効な.CSVファイルとしてGCSに書きたい値の辞書があります。私は改行で区切られたテキストファイルとして辞書を書き出すことができますが、辞書を有効な.CSVに変換する例は見つけられないようです。データフローのパイプライン内でcsvを生成する最良の方法を誰にでも提案できますか？この回答はquestionです。CSVファイルからの読み込みですが、実際にはCSVファイルへ