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
