apache-beam-io

0熱

1答えて

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

1熱

2答えて

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

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

0熱

1答えて

Mavenを使用しないApache Beamプログラムの実行

Apache Sparkランナーを使用した簡単なサンプルBeamプログラムを実行します。 1）ローカルのプログラムを正常にコンパイルすることができました。 2）MavenがインストールされていないQAボックスにJARファイルをプッシュしたい。 3）サンプルプログラムをコンパイルして実行するMavenコマンドの例を見ています。 4）Mavenをインストールせずにコードを実行する手順を教えてください。

0熱

1答えて

Apacheのビーム2.1.0は、私は（Google App Engineのから）のStackdriverログに次のエラーを取得するGAE

における非既存の機能にグアバ前提条件checkArgumentを呼び出します。 java.lang.NoSuchMethodError: com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;I)V at org.apache.beam.sdk.io.gcp.datastore.DatastoreV1$Read.

3熱

2答えて

無限のコレクションをGCSに書く

同じトピックについて多くの質問がありました。しかし、私はまだGCSへの書き込みに問題があります。私はpubsubから話題を読んで、これをGCSにプッシュしようとしています。私はthis linkを参照しました。しかし、最新のビームパッケージでIOChannelUtilsが見つかりませんでした。 PCollection<String> details = pipeline .apply

0熱

1答えて

BigtableIO与えられた接頭辞を持つキーを読む

私は、与えられた接頭辞を持つすべての行を読む最良の方法を探しています。 BigTableIO.ReadにはwithKeyRangeメソッドがありますが、開始キーと終了キーを指定する必要があります。接頭辞から読むことを指定する方法はありますか？

0熱

1答えて

Apache Beamでファイルをバイト[]として読み取るには？

私たちは現在、Cloud DataflowのコンセプトApache Beam Pipelineの実証に取り組んでいます。いくつかのファイル（テキストもカスタムバイナリ形式もあります）をGoogle Cloud Bucketに入れ、これらのファイルをbyte []として読み込み、フロー内でそれらを逆シリアル化したいとします。ただし、テキスト以外のファイルを読み取ることができるBeamソースは見つ

1熱

1答えて

DoFnのHTTPクライアント

DataFlow上で実行されているApache BeamパイプラインのDoFnを通じてPOSTリクエストを行いたいと思います。このため、私は、PoolingHttpClientConnectionManagerで設定されたHttpClosableClientをインスタンス化するクライアントを作成しました。しかし、私は処理する各要素のクライアントをインスタンス化します。すべての要素で使用される

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

1熱

1答えて

FileBasedSinkのタイプパラメータは何ですか？

FileBasedSinkをバージョン2.0.0から2.2.0に拡張するカスタムシンクを移行しています。クラスが変更され、2つの追加の型パラメータを追加しました：UserTとDestinationTを： @Experimental(value=FILESYSTEM) public abstract class FileBasedSink<UserT,DestinationT,OutputT>