How to fix data locality in Spark Streaming?

2
votes

Spark version: 1.6.2

Spark configuration:

  • executors: 18

  • memory per executor: 30g

  • cores x executor: 2

Kafka configuration:

  • partitions: 18

Context:

I have a Spark-streaming pipeline that is using the new Direct approach from Kafka instead of receivers. This pipeline receives data from kafka, transforms the data, and inserts to Hive. We release that the trigger used as action to insert the data has a locality level equal to RACK_LOCAL, instead of PROCESS_LOCAL or NODE_LOCAL. Moreover, the input size column shows network information instead of memory. I have set the config variable spark.locality.wait=30s to test if spark waits enough to select the best locality mode, but none improvements has been shown.

Considering that the kafka partitions = num partitions in spark are equal, I thought that data was processed in the same executor which execute the action at the end, but I am not sure about that watching the results.

Spark task screenshotDAG

Does someone the idea how to force the task to NODE_LOCAL or PROCESS_LOCAL??

1
hadoopapache-sparkspark-streaming
Is your kafka cluster colocated with your Spark cluster? - maasg
@maasg are in a different cluster, A cluster for Kafka, and other for Spark. In the middle of transform process I have a collect, so I cached the dataframe before that. But I understood that if I cached the dataframe before the action, when the action is triggered, the data is stored in memory. So, once the information is cached that should mean that is stored in the spark cluster memory. - user8237526
could you add the code? collect is never a good thing unless really necessary - maasg
The code is large. the reason of collect is the following: I have a dataframe which has column hours, and I collect to know with range of hours is in each partition. For example, If I have data from hour=13 and hour=14, I want to be aware which differents hours I have by partition in order to filter the dataframe in the different hours and insert in their respective Hive partitions with a ìnsert into` statement. You can see the DAG that I have attached to the question. - user8237526
Check this discussion: twitter.com/maasg/status/881867363821527040 - maasg

1 Answers

3
votes

As said in the comments, Locality is in reference to the source of the information being used. The KafkaDirect approach ends up using KafkaRDD which define locality in terms of the original Kafka Queue nodes. This means unless your Kafka machines are colocated with Spark executors you won't be able to have a better locality.

With the receiver approach it would have reported better locality but only because it basically was ignoring the Kafka -> Spark step. With receivers the data is being transferred then held on the Receiver's running node. This meant that Spark could report work on that data as "Process" or "Node" local since it can work on that transferred data without moivng it again but the same underlying first transfer transfer would still be happening.