問題タブ [data-ingestion]

REST API 呼び出しを使用してエンドポイントからデータを取り込み、そのデータを HDFS に保存する予定です。REST 呼び出しは定期的に (毎日または場合によっては 1 時間ごとに) 行われます。

私はすでに Flume を使用して Twitter の取り込みを行っていますが、Flume を使用することは私の現在のユースケースには適さないと思います。なぜなら、私は Twitter でこのような継続的なデータ ファイアホースを使用しておらず、個別の定期的な時間制限のある呼び出しを使用しているためです。

私が今考えているアイデアは、REST API 呼び出しを処理して HDFS に保存するカスタム Java を使用し、その Java jar で Oozie コーディネーターを使用することです。

設計と、このユースケースに使用する Hadoop ベースのコンポーネントについて、提案/代替案 (現在考えているものよりも簡単な場合) を聞きたいです。私が Flume に固執できると思われる場合は、その方法も教えてください。

私が見つけたすべての検索は、1 つのテーブルをインポートする方法、または import-all-tables を推奨する方法を示しています。データベースから 440 個のテーブルのうち 35 個が必要な場合はどうなりますか。1 つのコマンドを記述してテーブルをカンマで区切ることはできますか?それとも、スクリプトに入れてコマンドを何度もコピー アンド ペーストし、そのたびにテーブル名を変更する必要がありますか?

私がしたいこと：

私がしなければならないかもしれないと私が恐れていること：

これは私が投稿する最初の問題なので、情報や平凡な書式設定を見逃している場合はお詫びします. 必要に応じて更新できます。

できるだけ多くの詳細を追加しようとします。RDBMSデータをNeo4jのグラフノードと関係に変換する、あまり最適化されていないSparkジョブがあります。

これをする。これが私が従う手順です：

1. spark sql と結合を使用して、非正規化されたデータフレーム 'data' を作成します。

2. 「データ」の Foreach 行は、以下を行う graphInsert 関数を実行します。

   を。行の内容を読み取る
   b. c. neo4j サイファー クエリを作成します ( Mergeコマンドを使用して、都市が 1 つだけになるようにします。たとえば、シカゴが RDBMS テーブルに複数の行で表示される場合に、Neo4j でシカゴを作成します)
   c.
   d. neo4j に接続します 。クエリを実行します
   e. neo4j から切断する
   

これが私が直面している問題のリストです。

1. 挿入は遅いです。

マージ クエリは作成よりも遅いことはわかっていますが、レコードごとに接続および切断する代わりにこれを行う別の方法はありますか? これは私の最初のドラフト コードであり、1 つの接続を使用して、異なる Spark ワーカー ノード上の複数のスレッドから挿入する方法に苦労している可能性があります。したがって、レコードごとに接続および切断します。

2. ジョブはスケーラブルではありません。1コアでも問題なく動きます。2 つの Spark コアでジョブを実行するとすぐに、マージ クエリを実行している場合でも、同じ名前の 2 つの都市が突然表示されます。例: Merge の使用に違反しているシカゴの都市が 2 つあります。Merge は「存在しない場合は作成する」のような機能をしていると思います。

私の実装がneo4jの一部またはsparkで間違っているかどうかはわかりません。誰かがこれをより良い規模で実装するのに役立つドキュメントに私を導くことができれば、私はこの仕事のために最大限の可能性を利用する必要がある大きなスパーククラスターを持っているので役に立ちます.

アルゴリズムではなくコードに興味がある場合。以下は、scala での graphInsert の実装です。

sqoop で 6 つのマッパーを使用して Oracle からデータをインポートする場合、sqoop とソースの間に確立される接続の数。

1 つの接続になりますか、それとも各マッパーに対して 6 つの接続になります。

3 つのラックを備えたクラスターがあります。特定のソースのセットについては、特定のソースから他の宛先へのトラフィックを監視できるように、それらを 1 つのラックにのみダンプする必要があります。私の質問は簡単です。特定のデータ ソースを 1 つのラックに複製し、残りのソースを他のすべて (2 つまたは 3 つ) のラックに複製することは可能ですか?

ありがとう。

概要: 多重化シナリオがあり、静的に構成された値ではなく、フィールドの変数値 (日付など) に基づいて動的に多重化する方法を知りたいです。

詳細: entityId で区切られた入力があります。使用しているエンティティがわかっているので、典型的な Flume マルチチャネル選択で構成できます。

各チャネルは、個別の HDFSEventSink である「hdfsSink-n」に移動します。

これにより、エンティティごとにファイルが生成されますが、これで問題ありません。ここで、動的な 2 番目の変数、つまり日付を導入したいと思います。イベントの日付に応じて、エンティティごとに日付ごとにファイルを作成したいと考えています。日付は動的な値であるため、多数のシンクを事前に構成して、それぞれが別のファイルに送信されるようにすることはできません。また、シンクごとに 1 つの HDFS 出力のみを指定できます。

つまり、「Multiple Outputs HDFSEventSink」が必要だったようなものです (Hadoop の MultipleOutputs ライブラリと同様の方法で)。Flume にそのような機能はありますか?

そうでない場合、これを修正または回避するエレガントな方法はありますか? もう 1 つのオプションは、HDFSEventSink を変更することです。これは、イベントごとに「realName」(文字列) を別々に作成することで実装できるようです。

spark がドライバーを介してデータをプルするときと、spark がドライバーを介してデータをプルする必要がないときを簡単に説明しようとしています。

3 つの質問があります -

1. 20 TB のフラット ファイル ファイルが HDFS に保存されていて、ドライバー プログラムから、それぞれのライブラリのすぐに使える関数 (sc.textfile(path) またはsc.textfile(path).toDFなど) のいずれかを使用して、データ フレームまたは RDD にプルするとします。ドライバーが 32 GB のメモリのみで実行されている場合、ドライバー プログラムが OOM を持つようになりますか? または、少なくともドライバーのジムにスワップがありますか? それとも、ドライバーを経由せずにデータフレーム/RDD を作成するために、HDFS から Spark エグゼキューターにデータを配布するのに十分スマートな Spark と Hadoop でしょうか?
2. 外部 RDBMS からの場合を除いて、1 とまったく同じ質問ですか?
3. 特定のノードのファイル システム (Unix ファイル システムのみ、20 TB ファイルであり、HDFS ではない) を除いて、1 とまったく同じ質問ですか?