sparkからの外部データベースでのマイクロバッチルックアップ

Question

私はSparkを使用してログラインを処理する必要があります。処理のステップの1つは、外部DBの特定の値をルックアップすることです。

例： ログラインに複数のキーと値のペアが含まれています。ログに存在するキーの1つが「key1」です。ルックアップコールにはこのキーを使用する必要があります。 RDDの "key1"の各値に対して、外部DBで複数のルックアップを連続して実行したくありません.RDDにある "key1"のすべての値のリストを作成し、外部DBで単一参照呼び出しを行いたい。

ようになり、各ログ行からキーを抽出する私のコードは次のとおりです。

lines.foreachRDD{rdd => rdd.map(line => extractKey(line)) 
// next step is lookup 
// then further processing 

.MAP機能は、各ログ行のために呼ばれ、私はわからないことになるが、どのように私はのリストを作成することができます外部ルックアップに使用できるキー。

おかげ

Answer 1

あなたはこれをしたいように見えます：

lines.groupByKey().filter() 

あなたはより多くの情報を提供してもらえますか？

Answer 2

collectを使用してください。

lines.foreachRDD{rdd => 
    val keys = rdd.map(line => extractKey(line)).collect() 
    // here you can use keys List 

おそらくあなたもmapPartitionsを使用する必要があります：1つのパーティションあたり1回のコールがあります

lines.foreachRDD{rdd => 
    rdd.foreachPartition(iter => { 
     val keys = iter.map(line => extractKey(line)).toArray 
     // here you can use keys Array 

    } 
} 

、この方法は、あなたが `collect`または` collectAsListを使用することができ、シリアル化問題