I have a small Hive Table with 15 million rows saved over HDFS (parquet/1152 files - over 30GB).
I am doing LDA over scientific abstracts. Therefore, the first step is extracting some noun phrases/chunk phrases by using StanfordNLP which I wrote an UDF to achieve this goal.
Now performance wise, there are two scenarios which each has very different results.
Scenario 1:
val hiveTable = hivecontext.sql("""
SELECT ab AS text,
pmid AS id
FROM scientific.medline
LIMIT 15000000
""")
Then I call my UDF over my hiveTable:
val postagsDF = hiveTable.withColumn("words", StanfordNLP.posPhrases(col("text")))
Now if I trigger any action/transformation such as .count() or do CountVectorizer() on "postagsDF" I see 2 stages. One with the appropriate number of tasks (partitions) and the other stage with only one task. First one ends very fast after doing some Input/Shuffle writes but the second one with only one task takes a long time. It seems that my UDF is being executed in this stage which only has one task. (takes hours, no resource activity during its completion)
Scenario 2:
val hiveTable = hivecontext.sql("""
SELECT ab AS text,
pmid AS id
FROM scientific.medline
LIMIT 15000000
""")
I repartition my DataFrame to the exact number of partitions detected by spark based on the number of parquets. (I can choose any other number but the number seems OK since I have over 500 cores available - 2 tasks per core)
val repartitionedDocDF = docDF.repartition(1152)
Now calling my UDF over my hiveTable:
val postagsDF = hiveTable.withColumn("words", StanfordNLP.posPhrases(col("text")))
However, any action/transformation this time will be four stages. Two of the stages (let's say count) are 1152 tasks and two of them are single task. I can see my UDF is being executed in one of those stages with 1152 tasks by all the executors using my entire cluster properly.
Results of scenario number 1: Looking at my cluster there is not much going on during the long-running single-task stage. There is no cpu usage, no memory, no network and no IO activity. Just one executor with one task that is applying my UDF over each document/column.
Benchmark: Scenario number 1 takes 3-4 hours to finish only 1 million rows. (I couldn't wait to see how much it takes for 15 million rows)
Results of scenario number 2: Looking at my cluster I can clearly see all my resources are being utilised. All my nodes are almost at the full capacity.
Benchmark: Scenario number 2 takes over 30 minutes for 15 million rows.
Real questions
What just happened? I thought UDFs over Dataframe would run in parallel by default. Maybe do repartition if the number of partions/tasks are more or less than total number of cores, but at least parallel over the default 200 partitions/tasks. I just want to understand why the UDf in my case is single task and ignoring both the default 200 and actual partition size. (It is not just about performance, it's single task job vs multi tasks job)
Is there any other way to make UDF to be executed over all the executors in parallel without calling repartition. I have nothing against repartitioning, but it is very expensive operation which I don't think it should be the only way of making UDF runs in parallel. Even when I repartition to the exact same number of partitions/files I still have to watch over 20GB shuffle reads and writes fly over my cluster.
I have read everything about repartitioning and UDFs but I couldn't find similar issue which one can't run the UDF by default in parallel unless it does repartitioning. (simple UDF when you cast a type of a column from int to bigint might not be visible, but when you do NLP it really is visible)
My cluster size: 30 nodes (16core/32G) - Spark 1.6 Cloudera CDH 5.11.1
Spark: --driver-cores 5 --driver-memory 8g --executor-cores 5 --executor-memory 5g --num-executors 116
Many thanks,
UPDATE:
I ran the same code without the LIMIT clause and it did it in 18 minutes! So the LIMIT was the reason (more on this in the answer):
