spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Nicholas Chammas <nicholas.cham...@gmail.com>
Subject Re: Downloads from S3 exceedingly slow when running on spark-ec2
Date Sat, 20 Dec 2014 22:55:46 GMT
Is the operation slow every time or does it run normally if you repeat the
operation within the same app?

Nick


On Thu, Dec 18, 2014 at 8:56 AM, Jon Chase <jon.chase@gmail.com> wrote:

> I'm running a very simple Spark application that downloads files from S3,
> does a bit of mapping, then uploads new files.  Each file is roughly 2MB
> and is gzip'd.  I was running the same code on Amazon's EMR w/Spark and not
> having any download speed issues (Amazon's EMR provides a custom
> implementation of the s3n:// file system, FWIW).
>
> When I say exceedingly slow, I mean that it takes about 2 minutes to
> download and process a 2MB file (this was taking ~2 seconds on the same
> instance types in Amazon's EMR).  When I download the same file from the
> EC2 machine with wget or curl, it downloads in ~ 1 second.  I've also done
> other bandwidth checks for downloads from other external hosts - no speed
> problems there.
>
> Tried this w/Spark 1.1.0 and 1.1.1.
>
> When I do a thread dump on a worker, I typically see this a lot:
>
>
>
> "Executor task launch worker-7" daemon prio=10 tid=0x00007fd174039000
> nid=0x59e9 runnable [0x00007fd1f7dfb000]
>    java.lang.Thread.State: RUNNABLE
> at java.net.SocketInputStream.socketRead0(Native Method)
> at java.net.SocketInputStream.read(SocketInputStream.java:152)
> at java.net.SocketInputStream.read(SocketInputStream.java:122)
> at sun.security.ssl.InputRecord.readFully(InputRecord.java:442)
> at sun.security.ssl.InputRecord.read(InputRecord.java:480)
> at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:927)
> - locked <0x00000007e44dd140> (a java.lang.Object)
> at sun.security.ssl.SSLSocketImpl.readDataRecord(SSLSocketImpl.java:884)
> at sun.security.ssl.AppInputStream.read(AppInputStream.java:102)
> - locked <0x00000007e44e1350> (a sun.security.ssl.AppInputStream)
> at java.io.BufferedInputStream.fill(BufferedInputStream.java:235)
> at java.io.BufferedInputStream.read(BufferedInputStream.java:254)
> - locked <0x00000007e44ea800> (a java.io.BufferedInputStream)
> at org.apache.commons.httpclient.HttpParser.readRawLine(HttpParser.java:78)
> at org.apache.commons.httpclient.HttpParser.readLine(HttpParser.java:106)
> at
> org.apache.commons.httpclient.HttpConnection.readLine(HttpConnection.java:1116)
> at
> org.apache.commons.httpclient.MultiThreadedHttpConnectionManager$HttpConnectionAdapter.readLine(MultiThreadedHttpConnectionManager.java:1413)
> at
> org.apache.commons.httpclient.HttpMethodBase.readStatusLine(HttpMethodBase.java:1973)
> at
> org.apache.commons.httpclient.HttpMethodBase.readResponse(HttpMethodBase.java:1735)
> at
> org.apache.commons.httpclient.HttpMethodBase.execute(HttpMethodBase.java:1098)
> at
> org.apache.commons.httpclient.HttpMethodDirector.executeWithRetry(HttpMethodDirector.java:398)
> at
> org.apache.commons.httpclient.HttpMethodDirector.executeMethod(HttpMethodDirector.java:171)
> at
> org.apache.commons.httpclient.HttpClient.executeMethod(HttpClient.java:397)
> at
> org.apache.commons.httpclient.HttpClient.executeMethod(HttpClient.java:323)
> at
> org.jets3t.service.impl.rest.httpclient.RestS3Service.performRequest(RestS3Service.java:342)
> at
> org.jets3t.service.impl.rest.httpclient.RestS3Service.performRestHead(RestS3Service.java:718)
> at
> org.jets3t.service.impl.rest.httpclient.RestS3Service.getObjectImpl(RestS3Service.java:1599)
> at
> org.jets3t.service.impl.rest.httpclient.RestS3Service.getObjectDetailsImpl(RestS3Service.java:1535)
> at org.jets3t.service.S3Service.getObjectDetails(S3Service.java:1987)
> at org.jets3t.service.S3Service.getObjectDetails(S3Service.java:1332)
> at
> org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrieveMetadata(Jets3tNativeFileSystemStore.java:111)
> at sun.reflect.GeneratedMethodAccessor15.invoke(Unknown Source)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:606)
> at
> org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:82)
> at
> org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:59)
> at org.apache.hadoop.fs.s3native.$Proxy6.retrieveMetadata(Unknown Source)
> at
> org.apache.hadoop.fs.s3native.NativeS3FileSystem.getFileStatus(NativeS3FileSystem.java:330)
> at
> org.apache.hadoop.fs.s3native.NativeS3FileSystem.mkdir(NativeS3FileSystem.java:432)
> at
> org.apache.hadoop.fs.s3native.NativeS3FileSystem.mkdirs(NativeS3FileSystem.java:425)
> at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:1126)
> at
> org.apache.hadoop.mapred.FileOutputCommitter.getWorkPath(FileOutputCommitter.java:256)
> at
> org.apache.hadoop.mapred.FileOutputFormat.getTaskOutputPath(FileOutputFormat.java:244)
> at
> org.apache.hadoop.mapred.TextOutputFormat.getRecordWriter(TextOutputFormat.java:126)
> at
> org.apache.hadoop.mapred.lib.MultipleTextOutputFormat.getBaseRecordWriter(MultipleTextOutputFormat.java:44)
> at
> org.apache.hadoop.mapred.lib.MultipleOutputFormat$1.write(MultipleOutputFormat.java:99)
> at org.apache.spark.SparkHadoopWriter.write(SparkHadoopWriter.scala:94)
> at
> org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:986)
> at
> org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:974)
> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
> at org.apache.spark.scheduler.Task.run(Task.scala:54)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
>
>
>
>
> Here's my pseudo code:
>
> Launch cluster:
> ./spark-ec2 -k x -i x --wait=480 -m m3.xlarge -t m3.xlarge -s 2
> --spot-price=.1 -r eu-west-1 --master-opts=-Dspark.eventLog.enabled=true
> launch spark-ec2
>
>
> Run job (from the master):
> ~/spark/bin/spark-submit --class com.MyClass --master
> spark://ec2-xx-xx-xx-xx:7077 --deploy-mode client --driver-memory 6G
> --executor-memory 2G --conf spark.eventLog.enabled=true ~/my.jar --in-path
> s3n://bucket/path/prefix-* --out-path s3n://bucket/outpath
>
>
> Job (Java pseudo):
>
>     rdd=   ctx.textFile(cmd.inPath)
>                  .map(return parser.parse(line))
>                 .filter(return targetDate.equals(logLine.timestamp))
>                 .keyBy(return p.partitionKeyFor(logLine))
>
>
>             FileOutputFormat.setCompressOutput(jobConf, true);
>             FileOutputFormat.setOutputCompressorClass(jobConf,
> GzipCodec.class);
>
>
>       rdd.saveAsHadoopDataset(jobConf);
>
>
> Unfortunately, I haven't been able to get debugging turned up - I'm using
> slf4j/logback w/the commons-logging and log4j bridges.  Any pointers for
> getting that turned up to DEBUG would be helpful too.
>
>
> I've tried everything I can think of and am at my wit's end - any
> troubleshooting suggestions would be greatly appreciated!
>
>
>

Mime
View raw message