Spark-StreamingからElastic Searchへの出力のフィールド名のマッピング

Question

次のコードを使用して、Spark-Streamingの出力をElasticSearchに格納しています。私はスパークストリーミングの出力を適切な名前のi.e (Key, OsName, PlatFormName, Mobile, BrowserName, Count)にマッピングしたいと思います。しかし、現在わかるように、_1や_2などのようにESにマッピングされています。 さらに、いくつかのフィルタ、つまり(if PlatFormName = "ubuntu" then index the data)をESにインデックス付けする前に入れたいと思います。だから、どうすればいいの？ ElasticSearchで

val realTimeAgg = lines.map{ x => ((x.key, x.os, x.platform, x.mobile, x.browser), 1)}.reduceByKey(_+_) 

      val pageCounts = realTimeAgg.map  
      pageCounts.foreachRDD{ x => 
        if (x.toLocalIterator.nonEmpty) {  
         EsSpark.saveToEs(x, "spark/ElasticSearch") 
        } 
       } 

      ssc.start() 
      ssc.awaitTermination() 

出力：あなたが持つPairRDD（Tuple6、ロング）のデータ型を格納しているので、弾性検索文書の

{ 
      "_index": "spark", 
      "_type": "ElasticSearch", 
      "_id": "AVTH0JPgzgtrAOUg77qq", 
      "_score": 1, 
      "_source": { 
       "_1": { 
        "_3": "Amiga", 
        "_2": "AmigaOS 1.3", 
        "_6": "SeaMonkey", 
        "_1": "Usedcar", 
        "_4": 0, 
        "_5": 0 
       }, 
       "_2": 1013 
      } 
     } 

Answer 1

キー等_1、_2、です。

キーを保持するには、ケースクラスをキーとして使用する必要があります。

val realTimeAgg = lines.map{ x => (x, 1)}.reduceByKey(_+_) 

私は、オブジェクトxのクラスはケースクラスであり、あなたは（つまり2ケースクラスのインスタンスが等しいかどうかをチェックするための）削減を行うため、そのクラスのすべてのフィールドを使用することを想定しています。そのクラスのすべてのフィールドが平等のために使用するクラスの自然キーされていない場合、次の2つのオプションがある -

1. をオーバーライドは、あなたのケースクラス
2. のために等しいとhashCodeだけ持っている別のケースクラスを作成します。 （x.key、x.os、x.platform、x.mobile、x.browser）のタ​​プルで使用したフィールド）を取得し、最初の行のタプルではなくそのケースクラスにマッピングします。 map {x => ...}。

ElasticSearchに書き込む前に必要なフィルタを追加できます。

pageCounts.foreachRDD { x => 
         if (x.toLocalIterator.nonEmpty) { 
          val y = x.filter(z => z._1.platform == "ubuntu")  
          EsSpark.saveToEs(y, "spark/ElasticSearch") 
        } 
       } 

PS：あなたは私がlines.map（X =>（X、1））を示唆しているようにキーとして（ケースクラス、ロング）ケースクラスと対RDDをテストしている場合reduceByKey（_ +。 _）。 Spark Shellに関連したバグがあります。このクラスでは、操作クラスをキークラスとして正しく機能しません。jira issue