apache-beam

0熱

1答えて

Apache Beam - 複数のウィンドウを出力するウィンドウをスライドさせる

セッションのPCollectionを取得し、チャネル/接続ごとの平均セッション時間を取得しようとしています。私は、初期のトリガーが生成された各ウィンドウに対して発火するようなことをしています.60分のウィンドウが1分ごとにスライドすると、初期のトリガーは60回発火します。アウトプットのタイムスタンプを見ると、毎分60分のウィンドウがあります。私は最後のウィンドウのためにトリガーを1回発射したいので

0熱

2答えて

FixedWindowsとDurationでのValueProvider <Long>の使用

Apache Beam 2.xに移行するには、可能な限りテンプレートを使用し、それに応じてValueProviderを使用したいと考えています。私のロジックでは、FixedWindowを使用していますが、期間は柔軟なので、むしろValueProviderから取得します。 FixedWindows.of（）はDurationしか取得できないため、ValueProviderを取得してそこから期間を取

1熱

1答えて

Apache Beam：PTransformとPValue

任意のタイプXとYのPTransform<PCollection<X>, PCollection<Y>>を指定します。この例では正確に何が変換され、PValueは正確には何ですか？ PValueはグラフの最後の頂点を定義していますか？

0熱

1答えて

Apache Beam -BiasQueryIO Apexランナーを使用

apex/spark runnerを使用してApacheビームのテーブルにデータを書き込む。ただし、apexランナーを使用してプログラムを実行している間は例外が発生します。 List<TableFieldSchema> fields = new ArrayList<>(); fields.add(new TableFieldSchema().setName("Id").setType

0熱

1答えて

Apache beam KafkaIO外部データストアへのオフセット管理

私は、Apacheのビーム上でKafkaIOを使用して複数のkafkaブローカーから読もうとしています。オフセット管理のデフォルトオプションはkafkaパーティション自体です（kafka> 0.9のzookeperは使用しません）。この設定では、私はジョブ/パイプラインを再起動すると、重複しているレコードと不足しているレコードに問題があります。私が読んだことから、これを処理する最良の方法は、外部

0熱

1答えて

Apache Beam「mvn package」はEclipse経由で動作しません

Eclipse IDEと関連プラグインを使用している場合、IDEからローカルでもデータフローでも簡単にジョブを実行できます。ジョブを「プロダクト化」しようとすると（テンプレートを使用できない既知のDataflow/Bigqueryの制限のために）、「クラウドビルド」環境でコマンドラインから「mvn package」を使用しようとしています。それは常に "パッケージorg.apache.beam.

0熱

1答えて

データフロージョブが特定のdlに完了した後にメールをトリガーする方法

私は一定の時間間隔でトリガーするGoogleクラウドデータフロージョブを持っています。データフロージョブが完了した後で、ジョブのステータスを特定のメールIDに送信した後、メールをトリガーする必要があります。あなたはまた、result.metrics()によって返された電子メールに完成したジョブのメトリックを含めることができ PipelineResult result = pipeline.run(

1熱

2答えて

Apache Beamを使用してデータベースからバルクデータを読み取る

私のクエリが何百万行も返された場合、JdbcIOがどのようにクエリを並列に実行するかを知りたいと思います。私はhttps://issues.apache.org/jira/browse/BEAM-2803と関連するプルリクエストを参照しました。私はそれを完全に理解できませんでした。 ReadAllexpandの方法はParDoを使用します。したがって、データベースへの複数の接続を作成して、データ

0熱

1答えて

Dataflow 2.xは、PCollectionTuple.apply（）の呼び出し時に間違ったパラメータタイプについて不平を言います

私は既存のパイプラインをデータフロー2.xに移行しています。パイプラインの最終段階では、データはGoogle Cloud Servicesに書き込まれます。データは.gzに圧縮する必要があります。以前は（データフロー1.xの実装では）私たちはこれを行うために独自のシンクを作成していました。データフロー2.xでは、これを行うための組み込みの方法があります。正しいコードでなければならないものがあります

0熱

1答えて

BigQueryIO.Writeによってスローされた例外をキャッチして、出力に失敗したデータを救助する方法はありますか？

Cloud Pub/Subからデータを読み取り、それをCloud DataflowでBigQueryに書きたいとします。各データには、データ自体が保存されるテーブルIDが含まれています。のBigQueryへの書き込みに失敗した様々な要因があります。表のIDのフォーマットが間違っているが。データセットが存在しません。データセットでは、パイプラインにアクセスできません。ネットワーク障害。エ