Spark with a flow of kafka with preservation until a slow elasticity search result

Question

I have a list of data, the value is basically a bson document (think json), each json varies from 5k to 20k. It can either be in the format of a bson object, or directly converted to json:

Key, Value -------- K1, JSON1 K1, JSON2 K2, JSON3 K2, JSON4 

I expect groupByKey to produce:

 K1, (JSON1, JSON2) K2, (JSON3, JSON4) 

so that when executed:

 val data = [...].map(x => (x.Key, x.Value)) val groupedData = data.groupByKey() groupedData.foreachRDD { rdd => //the elements in the rdd here are not really grouped by the Key } 

I am so confused about the behavior of RDD. I read many articles on the Internet, including the official site from Spark: https://spark.apache.org/docs/0.9.1/scala-programming-guide.html

Still not able to achieve what I want.

-------- UPDATED ---------------------

Basically, I really need it to be grouped by key, the key is the index that will be used in Elasticsearch so that I can execute the batch process based on the key through Elasticsearch for Hadoop:

 EsSpark.saveToEs(rdd); 

I can not do for each section, because Elasticsearch accepts only RDD. I tried using sc.MakeRDD or sc.parallize, both say it is not serializable.

I tried using:

 EsSpark.saveToEs(rdd, Map( "es.resource.write" -> "{TheKeyFromTheObjectAbove}", "es.batch.size.bytes" -> "5000000") 

The configuration documentation is here: https://www.elastic.co/guide/en/elasticsearch/hadoop/current/configuration.html

But this is a VERY slow comparison with not using the configuration to determine the dynamic index based on the value of a single document, I suspect that it parses every json to dynamically extract the value.

Answer 1

Here is an example.

 import org.apache.spark.SparkConf import org.apache.spark.rdd.RDD import org.apache.spark.sql.SparkSession object Test extends App { val session: SparkSession = SparkSession .builder.appName("Example") .config(new SparkConf().setMaster("local[*]")) .getOrCreate() val sc = session.sparkContext import session.implicits._ case class Message(key: String, value: String) val input: Seq[Message] = Seq(Message("K1", "foo1"), Message("K1", "foo2"), Message("K2", "foo3"), Message("K2", "foo4")) val inputRdd: RDD[Message] = sc.parallelize(input) val intermediate: RDD[(String, String)] = inputRdd.map(x => (x.key, x.value)) intermediate.toDF().show() // +---+----+ // | _1| _2| // +---+----+ // | K1|foo1| // | K1|foo2| // | K2|foo3| // | K2|foo4| // +---+----+ val output: RDD[(String, List[String])] = intermediate.groupByKey().map(x => (x._1, x._2.toList)) output.toDF().show() // +---+------------+ // | _1| _2| // +---+------------+ // | K1|[foo1, foo2]| // | K2|[foo3, foo4]| // +---+------------+ output.foreachPartition(rdd => if (rdd.nonEmpty) { println(rdd.toList) }) // List((K1,List(foo1, foo2))) // List((K2,List(foo3, foo4))) }