問題タブ [camus]

3 ノードの EMR クラスターを使用して、kafka から s3 にデータを移動するために gobblin を実行しています。私はhadoop 2.6.0で実行しており、2.6.0に対してゴブリンも構築しました。

map-reduce ジョブが正常に実行されているようです。私のhdfsには、メトリックと作業ディレクトリが表示されます。metrics にはいくつかのファイルがありますが、作業ディレクトリは空です。S3 バケットには最終出力があるはずですが、データがありません。そして最後にこう言っている

出力タスクの状態パス /gooblinOutput/working/GobblinKafkaQuickStart_mapR3/output/job_GobblinKafkaQuickStart_mapR3_1460132596498 が存在しません 作業ディレクトリを削除しました /gooblinOutput/working/GobblinKafkaQuickStart_mapR3

最終ログは次のとおりです。

ここに私のconfファイルがあります：

ファイル 2 : kafka-to-s3.pull

コマンドの実行

何が起こっているのかわからない。誰か助けてくれませんか？

私は得ています

スレッド「メイン」の例外 java.lang.NoClassDefFoundError: com/linkedin/camus/etl/IEtlKey.

コマンドの実行時:

以下の例外が発生しています..

camus-example-0.1.0-SNAPSHOT-shaded.jarクラスパスに含めました。

何か不足している場合はお知らせください。

前もって感謝します

スミャジット

Kafka (ver 0.82) から avro ペイロードを取得し、HDFS に .avro ファイルとして保存するために、約 1 年間 Camus を実行してきました。いくつかの Kafka トピックを使用しています。最近、社内の新しいチームが、運用前環境で約 60 の新しいトピックを登録し、これらのトピックにデータを送信し始めました。チームは、データを kafka トピックにルーティングするときにいくつかの間違いを犯しました。これにより、Camus がこれらのトピックの avro にペイロードを逆シリアル化したときにエラーが発生しました。「その他の失敗」エラーしきい値を超えたため、Camus ジョブは失敗しました。失敗後の Camus の動作は驚くべきものでした。他の開発者に確認して、観察した動作が予期されたものなのか、それとも実装に問題があるのか​​を確認したかったのです。

「その他の失敗」しきい値を超えたために Camus ジョブが失敗したときに、この動作に気付きました。 1. すべてのマッパー タスクが成功したため、TaskAttempt のコミットが許可されました。最終的な HDFS の場所。2. CamusJob は、% エラー率を計算するときに例外をスローし (これはマッパー コミットに続いています)、ジョブが失敗しました 3. ジョブが失敗したため (私が思うに)、Kafka オフセットは進みませんでした

この動作で遭遇した問題は、Camus ジョブが 5 分ごとに実行されるように設定されていることです。そのため、データが HDFS にコミットされ、ジョブが失敗し、Kafka オフセットが更新されていないことを 5 分ごとに確認しました。これは、ディスクがいっぱいになるまで重複データを書き込んだことを意味します。

結果を確認する統合テストを作成しました。トピックに 10 個の適切なレコードを送信し、同じトピックに予期しないスキーマを使用する 10 個のレコードを送信し、そのトピックのみをホワイトリストに登録して Camus ジョブを実行し、10 個のレコードがHDFS に書き込まれ、Kafka オフセットは高度ではありません。以下は、そのテストのログのスニペットと、ジョブの実行中に使用したプロパティです。

これが Camus の予期された動作なのか、実装に問題があるのか​​、この動作 (データの複製) を防ぐための最善の方法は何かわかりません。

ありがとう〜マット

テストの CamusJob プロパティ:

マッパーが成功した後のコミット動作と、「その他」のしきい値を超えたために後続のジョブが失敗したことを示す、テストのログ スニペット:

Kafka から HDFS にメッセージを保存するための camus または gobblin の設定を手伝ってくれませんか。実際の例は素晴らしいかもしれません。

ゴブリンはまだインキュベーション段階にあり、カミュは段階的に廃止されています。したがって、どちらを使用するのが好ましいですか。

gobblin をダウンロードして gobblin-standalone.sh start を実行しましたが、エラーが発生しました

環境変数 GOBBLIN_JOB_CONFIG_DIR が設定されていません! .

私のプロジェクトでは、camus は長い間使用されており、更新されることはありません。camus プロジェクトは、kafka バージョン 0.8.2.2 を使用します。kafka 1.0.0 を使用するための回避策を見つけたいです。

そのため、ディレクトリを複製して依存関係を更新しました。これを行うと、 Message here には追加のパラメーターhereが必要です。

上記の github リンクに示されているように、コードはコンパイルされますが、ここの条件により、メッセージは kafka から読み取られません。

適切なデータコンストラクターとともにkafka依存関係を更新して機能させることは可能ですかkafka.message.Message?