Using options of S3FileSystem in ParquetSink to pass s3-credentials t…

…o ParquetWriter #373
51zero · Feb 28, 2018 · 598421c · 598421c
1 parent 7cd7d3f
commit 598421c
Show file tree

Hide file tree

Showing 4 changed files with 6 additions and 4 deletions.
diff --git a/eel-core/src/main/scala/io/eels/component/parquet/ParquetSink.scala b/eel-core/src/main/scala/io/eels/component/parquet/ParquetSink.scala
@@ -41,7 +41,7 @@ case class ParquetSink(path: Path, options: ParquetWriteOptions = ParquetWriteOp
     if (options.overwrite && fs.exists(path))
       fs.delete(path, false)
 
-    val writer = RowParquetWriterFn(path, schema, options.metadata, options.dictionary, options.roundingMode)
+    val writer = RowParquetWriterFn(path, schema, options.metadata, options.dictionary, options.roundingMode, fs.getConf)
 
     override def write(row: Row): Unit = {
       writer.write(row)

diff --git a/eel-core/src/main/scala/io/eels/component/parquet/RowParquetWriterFn.scala b/eel-core/src/main/scala/io/eels/component/parquet/RowParquetWriterFn.scala
@@ -35,7 +35,8 @@ object RowParquetWriterFn {
             schema: StructType,
             metadata: Map[String, String],
             dictionary: Boolean,
-            roundingMode: RoundingMode): ParquetWriter[Row] = {
+            roundingMode: RoundingMode,
+            fsConfig: Configuration): ParquetWriter[Row] = {
     val config = ParquetWriterConfig()
     val messageType = ParquetSchemaFns.toParquetMessageType(schema)
     new RowParquetWriterBuilder(path, messageType, roundingMode, metadata)
@@ -46,6 +47,7 @@ object RowParquetWriterFn {
       .withValidation(config.validating)
       .withWriteMode(ParquetFileWriter.Mode.CREATE)
       .withWriterVersion(ParquetProperties.WriterVersion.PARQUET_1_0)
+      .withConf(fsConfig)
       .build()
   }
 }
diff --git a/eel-core/src/test/scala/io/eels/component/parquet/DecimalWriterTest.scala b/eel-core/src/test/scala/io/eels/component/parquet/DecimalWriterTest.scala
@@ -29,7 +29,7 @@ class DecimalWriterTest extends FunSuite {
     if (fileSystem.exists(path)) fileSystem.delete(path, false)
 
     // Write out the decimal values
-    val parquetWriter = RowParquetWriterFn(path = path, schema = schema, metadata = Map.empty, dictionary = false, roundingMode = RoundingMode.UP)
+    val parquetWriter = RowParquetWriterFn(path = path, schema = schema, metadata = Map.empty, dictionary = false, roundingMode = RoundingMode.UP, fileSystem.getConf)
     expectedBigDecimals.foreach { expectedBigDecimal =>
       println(s"Writing row with value $expectedBigDecimal")
       parquetWriter.write(Row.fromMap(schema, Map("bd" -> expectedBigDecimal)))

diff --git a/eel-hive/src/main/scala/io/eels/component/hive/dialect/ParquetHiveDialect.scala b/eel-hive/src/main/scala/io/eels/component/hive/dialect/ParquetHiveDialect.scala
@@ -66,7 +66,7 @@ case class ParquetHiveDialect(options: ParquetWriteOptions = ParquetWriteOptions
 
       private val _records = new AtomicInteger(0)
       logger.debug(s"Creating parquet writer at $path")
-      private val writer = RowParquetWriterFn(path, schema, metadata, true, roundingMode)
+      private val writer = RowParquetWriterFn(path, schema, metadata, true, roundingMode, fs.getConf)
 
       override def write(row: Row) {
         require(row.values.nonEmpty, "Attempting to write an empty row")