Jyotirmoy, would you like to open up a PR with your changes that you did to get S3 reading working?
On Thu, Jul 6, 2017 at 1:26 PM, Jyotirmoy Sundi <[email protected]> wrote: > Hi Ted, > BEAM-2500 is for reading it seems, I made a couple of changes in > beam HadoopFileSystem > and able to read s3 data but write i am still facing the issues above. any > help would highly appreciate. > > On Wed, Jul 5, 2017 at 8:49 PM, Ted Yu <[email protected]> wrote: > > > Please take a look at BEAM-2500 (and related JIRAs). > > > > Cheers > > > > On Wed, Jul 5, 2017 at 8:00 PM, Jyotirmoy Sundi <[email protected]> > > wrote: > > > > > Hi Folks, > > > > > > I am trying to write to s3 from beam. > > > > > > These are configs I am passing > > > > > > --hdfsConfiguration='[{"fs.default.name": "s3://xxx-output", > > > "fs.s3.awsAccessKeyId" :"xxx", "fs.s3.awsSecretAccessKey":"yyy"}]' > > > --input="/home/hadoop/data" --output="s3://xx-output/beam-output/" > > > > > > *Any idea how can I write to s3, I am using beam release-2.0.0* > > > > > > *Trace* > > > > > > 17/07/06 02:55:46 WARN TaskSetManager: Lost task 7.0 in stage 2.0 (TID > > 31, > > > ip-10-130-237-28.vpc.internal): org.apache.beam.sdk.util. > > > UserCodeException: > > > java.lang.NullPointerException > > > > > > at > > > org.apache.beam.sdk.util.UserCodeException.wrap( > > UserCodeException.java:36) > > > > > > at > > > org.apache.beam.sdk.io.WriteFiles$WriteShardedBundles$auxiliary$ > > > TXDiaduA.invokeProcessElement(Unknown > > > Source) > > > > > > at > > > org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement( > > > SimpleDoFnRunner.java:197) > > > > > > at > > > org.apache.beam.runners.core.SimpleDoFnRunner.processElement( > > > SimpleDoFnRunner.java:155) > > > > > > at > > > org.apache.beam.runners.spark.translation.DoFnRunnerWithMetrics. > > > processElement(DoFnRunnerWithMetrics.java:64) > > > > > > at > > > org.apache.beam.runners.spark.translation.SparkProcessContext$ > > > ProcCtxtIterator.computeNext(SparkProcessContext.java:165) > > > > > > at > > > org.apache.beam.runners.spark.repackaged.com.google.common. > > > collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:145) > > > > > > at > > > org.apache.beam.runners.spark.repackaged.com.google.common. > > > collect.AbstractIterator.hasNext(AbstractIterator.java:140) > > > > > > at > > > scala.collection.convert.Wrappers$JIteratorWrapper. > > > hasNext(Wrappers.scala:41) > > > > > > at scala.collection.Iterator$$anon$14.hasNext(Iterator.scala:388) > > > > > > at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) > > > > > > at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) > > > > > > at scala.collection.Iterator$class.foreach(Iterator.scala:727) > > > > > > at scala.collection.AbstractIterator.foreach(Iterator.scala:1157) > > > > > > at scala.collection.generic.Growable$class.$plus$plus$eq( > > > Growable.scala:48) > > > > > > at scala.collection.mutable.ArrayBuffer.$plus$plus$eq( > > > ArrayBuffer.scala:103) > > > > > > at scala.collection.mutable.ArrayBuffer.$plus$plus$eq( > > > ArrayBuffer.scala:47) > > > > > > at scala.collection.TraversableOnce$class.to( > TraversableOnce.scala:273) > > > > > > at scala.collection.AbstractIterator.to(Iterator.scala:1157) > > > > > > at > > > scala.collection.TraversableOnce$class.toBuffer(TraversableOnce. > > scala:265) > > > > > > at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157) > > > > > > at scala.collection.TraversableOnce$class.toArray( > > > TraversableOnce.scala:252) > > > > > > at scala.collection.AbstractIterator.toArray(Iterator.scala:1157) > > > > > > at > > > org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12. > > > apply(RDD.scala:927) > > > > > > at > > > org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12. > > > apply(RDD.scala:927) > > > > > > at > > > org.apache.spark.SparkContext$$anonfun$runJob$5.apply( > > > SparkContext.scala:1858) > > > > > > at > > > org.apache.spark.SparkContext$$anonfun$runJob$5.apply( > > > SparkContext.scala:1858) > > > > > > at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) > > > > > > at org.apache.spark.scheduler.Task.run(Task.scala:89) > > > > > > at org.apache.spark.executor.Executor$TaskRunner.run( > Executor.scala:227) > > > > > > 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) > > > > > > Caused by: java.lang.NullPointerException > > > > > > at java.io.File.<init>(File.java:277) > > > > > > at > > > org.apache.hadoop.fs.s3.S3OutputStream.newBackupFile( > > > S3OutputStream.java:92) > > > > > > at org.apache.hadoop.fs.s3.S3OutputStream.<init>( > S3OutputStream.java:84) > > > > > > at org.apache.hadoop.fs.s3.S3FileSystem.create(S3FileSystem.java:252) > > > > > > at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:915) > > > > > > at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:896) > > > > > > at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:793) > > > > > > at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:782) > > > > > > at > > > org.apache.beam.sdk.io.hdfs.HadoopFileSystem.create( > > > HadoopFileSystem.java:103) > > > > > > at > > > org.apache.beam.sdk.io.hdfs.HadoopFileSystem.create( > > > HadoopFileSystem.java:67) > > > > > > at org.apache.beam.sdk.io.FileSystems.create(FileSystems.java:207) > > > > > > at org.apache.beam.sdk.io.FileSystems.create(FileSystems.java:194) > > > > > > at org.apache.beam.sdk.io.FileBasedSink$Writer.open( > > > FileBasedSink.java:876) > > > > > > at > > > org.apache.beam.sdk.io.FileBasedSink$Writer. > > openUnwindowed(FileBasedSink. > > > java:842) > > > > > > at > > > org.apache.beam.sdk.io.WriteFiles$WriteShardedBundles. > > > processElement(WriteFiles.java:362) > > > > > > > > > > > > -- > > > Best Regards, > > > Jyotirmoy Sundi > > > > > > > > > -- > Best Regards, > Jyotirmoy Sundi >
