[
https://issues.apache.org/jira/browse/HUDI-1013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17140870#comment-17140870
]
sivabalan narayanan commented on HUDI-1013:
-------------------------------------------
[~uditme]: Great. Sure, happy to work with you folks. I have updated the
description w/ our plans on this effort. Here is the update on where we stand
for now.
I have some initial draft for end to end Bulk insert w/o converting to rdd. But
I had to do a rdd conversion for WriteStatus (output of mapPartitions call).
With this, we do see our perf has improved from existing bulk insert, but not
close to direct parquet datasource write in spark.
[Here|https://github.com/nsivabalan/hudi/commits/BulkInsertDatasetRows] is the
branch I have my updates so far and
[this|[https://github.com/nsivabalan/hudi/commit/b70fe56d4c0648ffc2ed5111c910a7580af2ea63]]
commit should have all the changes I have done so far.
With current scheme of things, here are some perf numbers.
500k records, 100 parallelism. 10 iterations.
|Benchmark |(iterationIndex)| Mode |Cnt |Score|Error|Units|
|benchmarkDirectParquetWrites|itr1|avgt|10|11.700 |± 4.587 |s/op|
|benchmarkBulkInsertRows|itr1|avgt|10|20.438|± 1.88|s/op|
Here, benchmarkDirectParquetWrites is direct parquet data source write in
spark, which is (2) as per the description.
Once [~dongwook] has taken a look at this and has a reasonable understanding on
what we plan to do and where we stand as of now, we can talk about further
steps.
> Bulk Insert w/o converting to RDD
> ---------------------------------
>
> Key: HUDI-1013
> URL: https://issues.apache.org/jira/browse/HUDI-1013
> Project: Apache Hudi
> Issue Type: Improvement
> Components: Writer Core
> Reporter: sivabalan narayanan
> Priority: Blocker
> Fix For: 0.6.0
>
>
> Our bulk insert(not just bulk insert, all operations infact) does dataset to
> rdd conversion in HoodieSparkSqlWriter and our HoodieClient deals with
> JavaRDD<HoodieRecord>s. We are trying to see if we can improve our
> performance by avoiding the rdd conversion. We will first start off w/ bulk
> insert and get end to end working before we decide if we wanna do this for
> other operations too after doing some perf analysis.
>
> On a high level, this is the idea
> 1. Dataset<Row> will be passed in all the way from spark sql writer to the
> storage writer. We do not convert to HoodieRecord at any point in time.
> 2. We need to use
> [ParquetWriteSupport|[https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala]]
> to write to Parquet as InternalRows.
> 3. So, gist of what we wanna do is, with the Dataset<Rows>s, sort by
> partition path and record keys, repartition by parallelism config, and do
> mapPartitions. Within MapPartitions, we will iterate through the Rows, encode
> to InternalRows and write to Parquet using the write support linked above.
> We first wanted to check if our strategy will actually improve the perf. So,
> I did a quick hack of just the mapPartition func in HoodieSparkSqlWriter just
> to see how the numbers look like. Check for operation
> "bulk_insert_direct_parquet_write_support"
> [here|#diff-5317f4121df875e406876f9f0f012fac]].
> These are the numbers I got. (1) is existing hoodie bulk insert. (2) is
> writing directly to parquet in spark. Code given below. (3) is the modified
> hoodie code i.e. operation bulk_insert_direct_parquet_write_support)
>
> | |5M records 100 parallelism input size 2.5 GB|
> |(1) Orig hoodie(unmodified)|169 secs. output size 2.7 GB|
> |(2) Parquet |62 secs. output size 2.5 GB|
> |(3) Modified hudi code. Direct Parquet Write |73 secs. output size 2.5 GB|
>
> So, essentially our existing code for bulk insert is > 2x that of parquet.
> Our modified hudi code (i.e. operation
> bulk_insert_direct_parquet_write_support) is close to direct Parquet write in
> spark, which shows that our strategy should work.
> // This is the Parquet write in spark. (2) above.
> transformedDF.sort(*"partition"*, *"key"*)
> .coalesce(parallelism)
> .write.format(*"parquet"*)
> .partitionBy(*"partition"*)
> .mode(saveMode)
> .save(*s"**$*outputPath*/**$*format*"*)
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
