Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[#781] Add support for Spark DataFrameWriter maxRecordsPerFile option #1017

Closed
wants to merge 7 commits into from
Closed

[#781] Add support for Spark DataFrameWriter maxRecordsPerFile option #1017

Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5113327
Add support for Spark DataFrameWriter maxRecordsPerFile and timeZone …
Mar 22, 2022
cc0d310
Resolve scala 2.13 compilation error
Mar 22, 2022
020c165
Remove support for timeZone option
Mar 24, 2022
a47338d
Add unit test
Mar 24, 2022
c9d092f
Add test cases as suggested
Mar 30, 2022
be14b5a
Merge remote-tracking branch 'upstream/master' into data-frame-writer…
Mar 30, 2022
d548e40
Need to add DeltaSQLCommandTest trait to DeltaOptionSuite
Mar 30, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 14 additions & 3 deletions core/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ import org.apache.spark.SparkException
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, FileFormatWriter, WriteJobStatsTracker}
import org.apache.spark.sql.functions.to_json
Expand Down Expand Up @@ -201,9 +202,8 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl

def writeFiles(
data: Dataset[_],
writeOptions: Option[DeltaOptions],
additionalConstraints: Seq[Constraint]): Seq[FileAction] = {
writeFiles(data, additionalConstraints)
writeFiles(data, None, additionalConstraints)
}

def writeFiles(
Expand All @@ -222,6 +222,7 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl
*/
def writeFiles(
data: Dataset[_],
writeOptions: Option[DeltaOptions],
additionalConstraints: Seq[Constraint]): Seq[FileAction] = {
if (DeltaConfigs.CHANGE_DATA_FEED.fromMetaData(metadata)) {
throw DeltaErrors.cdcWriteNotAllowedInThisVersion()
Expand Down Expand Up @@ -295,6 +296,16 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl
statsTrackers.append(basicWriteJobStatsTracker)
}

// Retain only known Spark writer options to avoid any potential compatibility issues
val options = writeOptions match {
case None => Map.empty[String, String]
case Some(writeOptions) =>
writeOptions.options.filterKeys(key =>
key.equalsIgnoreCase("maxRecordsPerFile") ||
key.equalsIgnoreCase(DateTimeUtils.TIMEZONE_OPTION)
zsxwing marked this conversation as resolved.
Show resolved Hide resolved
).toMap
}

try {
FileFormatWriter.write(
sparkSession = spark,
Expand All @@ -309,7 +320,7 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl
partitionColumns = partitioningColumns,
bucketSpec = None,
statsTrackers = optionalStatsTracker.toSeq ++ statsTrackers,
options = Map.empty)
options = options)
} catch {
case s: SparkException =>
// Pull an InvariantViolationException up to the top level if it was the root cause.
Expand Down