Spark data incorrect when more than 200 tasks
I'm fairly new to Spark.
We have data in avro files on hdfs.
We are trying to load up all the avro files (28 gigs worth right now) and do
an aggregation.
When we have less than 200 tasks the data all runs and produces the proper
results. If there are more than 200 tasks (as stated in the logs by the
TaskSetManager) the data seems to only group when it reads in the RDD from
hdfs by the first record in the avro file.
If I set: spark.shuffle.sort.bypassMergeThreshold greater than 200 data
seems to work. I don't understand why or how?
Here is the relevant code pieces:
JavaSparkContext context = new JavaSparkContext(
new SparkConf()
.setAppName(AnalyticsJob.class.getSimpleName())
.set("spark.serializer",
"org.apache.spark.serializer.KryoSerializer")
);
context.hadoopConfiguration().set("mapreduce.input.fileinputformat.input.dir.recursive",
"true");
context.hadoopConfiguration().set("mapreduce.input.fileinputformat.inputdir",
job.inputDirectory);
JavaRDD events = ((JavaRDD>)
context.newAPIHadoopRDD(
context.hadoopConfiguration(),
AvroKeyInputFormat.class,
AvroKey.class,
NullWritable.class
).keys())
.map(event -> event.datum())
.filter(key -> { return
Optional.ofNullable(key.getStepEventKey()).isPresent(); })
.mapToPair(event -> new Tuple2(event, 1))
.groupByKey()
.map(tuple -> tuple._1());
events.persist(StorageLevel.MEMORY_AND_DISK_2());
If I do a collect on events at this point the data is not as expected and
jumbled, so when we pass it onto the next job in our pipeline for
aggregation, the data doesn't come out as expected.
The downstream tasks maps to pairs again and stores in the db.
Thanks in advance for this help.
--
View this message in context:
http://apache-spark-user-list.1001560.n3.nabble.com/Spark-data-incorrect-when-more-than-200-tasks-tp21710.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
-
To unsubscribe, e-mail: user-unsubscr...@spark.apache.org
For additional commands, e-mail: user-h...@spark.apache.org
Spark Performance on Yarn
I'm a bit new to Spark, but had a question on performance. I suspect a lot of
my issue is due to tuning and parameters. I have a Hive external table on
this data and to run queries against it runs in minutes
The Job:
+ 40gb of avro events on HDFS (100 million+ avro events)
+ Read in the files from HDFS and dedupe events by key (mapToPair then a
reduceByKey)
+ RDD returned and persisted (disk and memory)
+ Then passed to a job that take the RDD and mapToPair of new object data
and then reduceByKey and foreachpartion do work
The issue:
When I run this on my environment on Yarn this takes 20+ hours. Running on
yarn we see the first stage runs to do build the RDD deduped, but then when
the next stage starts, things fail and data is lost. This results in stage 0
starting over and over and just dragging it out.
Errors I see in the driver logs:
ERROR cluster.YarnClientClusterScheduler: Lost executor 1 on X: remote
Akka client disassociated
15/02/20 00:27:36 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 3.1
(TID 1335,): FetchFailed(BlockManagerId(3, i, 33958), shuffleId=1,
mapId=162, reduceId=0, message=
org.apache.spark.shuffle.FetchFailedException: Failed to connect
toX/X:33958
Also we see this, but I'm suspecting this is because the previous stage
fails and the next one starts:
org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output
location for shuffle 1
Cluster:
5 machines, each 2 core , 8gb machines
Spark-submit command:
spark-submit --class com.myco.SparkJob \
--master yarn \
/tmp/sparkjob.jar \
Any thoughts or where to look or how to start approaching this problem or
more data points to present.
Thanks..
Code for the job:
JavaRDD events = ((JavaRDD>)
context.newAPIHadoopRDD(
context.hadoopConfiguration(),
AvroKeyInputFormat.class,
AvroKey.class,
NullWritable.class
).keys())
.map(event -> AnalyticsEvent.newBuilder(event.datum()).build())
.filter(key -> { return
Optional.ofNullable(key.getStepEventKey()).isPresent(); })
.mapToPair(event -> new Tuple2(event, 1))
.reduceByKey((analyticsEvent1, analyticsEvent2) -> analyticsEvent1)
.map(tuple -> tuple._1());
events.persist(StorageLevel.MEMORY_AND_DISK_2());
events.mapToPair(event -> {
return new Tuple2(
keySelector.select(event),
new RunningAggregates(
Optional.ofNullable(event.getVisitors()).orElse(0L),
Optional.ofNullable(event.getImpressions()).orElse(0L),
Optional.ofNullable(event.getAmount()).orElse(0.0D),
Optional.ofNullable(event.getAmountSumOfSquares()).orElse(0.0D)));
})
.reduceByKey((left, right) -> { return left.add(right); })
.foreachpartition(dostuff)
--
View this message in context:
http://apache-spark-user-list.1001560.n3.nabble.com/Spark-Performance-on-Yarn-tp21729.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
-
To unsubscribe, e-mail: user-unsubscr...@spark.apache.org
For additional commands, e-mail: user-h...@spark.apache.org
Re: Spark Performance on Yarn
A bit more context on this issue. From the container logs on the executor Given my cluster specs above what would be appropriate parameters to pass into : --num-executors --num-cores --executor-memory I had tried it with --executor-memory 2500MB 015-02-20 06:50:09,056 WARN org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl: Container [pid=23320,containerID=container_1423083596644_0238_01_004160] is running beyond physical memory limits. Current usage: 2.8 GB of 2.7 GB physical memory used; 4.4 GB of 5.8 GB virtual memory used. Killing container. Dump of the process-tree for container_1423083596644_0238_01_004160 : |- PID PPID PGRPID SESSID CMD_NAME USER_MODE_TIME(MILLIS) SYSTEM_TIME(MILLIS) VMEM_USAGE(BYTES) RSSMEM_USAGE(PAGES) FULL_CMD_LINE |- 23320 23318 23320 23320 (bash) 0 0 108650496 305 /bin/bash -c /usr/java/latest/bin/java -server -XX:OnOutOfMemoryError='kill %p' -Xms2400m -Xmx2400m -Djava.io.tmpdir=/dfs/yarn/nm/usercache/root/appcache/application_1423083596644_0238/container_1423083596644_0238_01_004160/tmp -Dspark.yarn.app.container.log.dir=/var/log/hadoop-yarn/container/application_1423083596644_0238/container_1423083596644_0238_01_004160 org.apache.spark.executor.CoarseGrainedExecutorBackend akka.tcp://sparkDriver@ip-10-168-86-13.ec2.internal:42535/user/CoarseGrainedScheduler 8 ip-10-99-162-56.ec2.internal 1 application_1423083596644_0238 1> /var/log/hadoop-yarn/container/application_1423083596644_0238/container_1423083596644_0238_01_004160/stdout 2> /var/log/hadoop-yarn/container/application_1423083596644_0238/container_1423083596644_0238_01_004160/stderr |- 23323 23320 23320 23320 (java) 922271 12263 461976 724218 /usr/java/latest/bin/java -server -XX:OnOutOfMemoryError=kill %p -Xms2400m -Xmx2400m -Djava.io.tmpdir=/dfs/yarn/nm/usercache/root/appcache/application_1423083596644_0238/container_1423083596644_0238_01_004160/tmp -Dspark.yarn.app.container.log.dir=/var/log/hadoop-yarn/container/application_1423083596644_0238/container_1423083596644_0238_01_004160 org.apache.spark.executor.CoarseGrainedExecutorBackend akka.tcp://sparkDriver@ip-10-168-86-13.ec2.internal:42535/user/Coarse -- View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Spark-Performance-on-Yarn-tp21729p21739.html Sent from the Apache Spark User List mailing list archive at Nabble.com. - To unsubscribe, e-mail: user-unsubscr...@spark.apache.org For additional commands, e-mail: user-h...@spark.apache.org
Deduping events using Spark
I'm still a bit new to Spark and am struggilng to figure out the best way to
Dedupe my events.
I load my Avro files from HDFS and then I want to dedupe events that have
the same nonce.
For example my code so far:
JavaRDD events = ((JavaRDD>)
context.newAPIHadoopRDD(
context.hadoopConfiguration(),
AvroKeyInputFormat.class,
AvroKey.class,
NullWritable.class
).keys())
.map(event -> AnalyticsEvent.newBuilder(event.datum()).build())
.filter(key -> { return
Optional.ofNullable(key.getStepEventKey()).isPresent(); })
Now I want to get back an RDD of AnalyticsEvents that are unique. So I
basically want to do:
if AnalyticsEvent.getNonce() == AnalyticsEvent2.getNonce() only return 1 of
them.
I'm not sure how to do this? If I do reduceByKey it reduces by
AnalyticsEvent not by the values inside?
Any guidance would be much appreciated how I can walk this list of events
and only return a filtered version of unique nocnes.
--
View this message in context:
http://apache-spark-user-list.1001560.n3.nabble.com/Deduping-events-using-Spark-tp23153.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
-
To unsubscribe, e-mail: user-unsubscr...@spark.apache.org
For additional commands, e-mail: user-h...@spark.apache.org
