WriteIntoDelta Command¶

WriteIntoDelta is a Delta command that can write data(frame) transactionally into a delta table.

WriteIntoDelta is a RunnableCommand (Spark SQL) logical operator.

Creating Instance¶

WriteIntoDelta takes the following to be created:

• DeltaLog
• SaveMode
• DeltaOptions
• Names of the partition columns
• Configuration
• Data (DataFrame)

WriteIntoDelta is created when:

• DeltaLog is requested to create an insertable HadoopFsRelation (when DeltaDataSource is requested to create a relation as a CreatableRelationProvider or a RelationProvider)
• DeltaCatalog is requested to createDeltaTable
• WriteIntoDeltaBuilder is requested to buildForV1Write
• CreateDeltaTableCommand command is executed
• DeltaDataSource is requested to create a relation (for writing) (as a CreatableRelationProvider)

ImplicitMetadataOperation¶

WriteIntoDelta is an operation that can update metadata (schema and partitioning) of the delta table.

Executing Command¶

run(
  sparkSession: SparkSession): Seq[Row]

run is part of the RunnableCommand (Spark SQL) abstraction.

run requests the DeltaLog to start a new transaction.

run writes and requests the OptimisticTransaction to commit (with DeltaOperations.Write operation with the SaveMode, partition columns, replaceWhere and userMetadata).

write¶

write(
  txn: OptimisticTransaction,
  sparkSession: SparkSession): Seq[Action]

write checks out whether the write operation is to a delta table that already exists. If so (i.e. the readVersion of the transaction is above -1), write branches per the SaveMode:

• For ErrorIfExists, write throws an AnalysisException.

  [path] already exists.
  

• For Ignore, write does nothing and returns back with no Actions.

• For Overwrite, write requests the DeltaLog to assert being removable

write updateMetadata (with rearrangeOnly option).

write...FIXME

write is used when:

• CreateDeltaTableCommand is executed
• WriteIntoDelta is executed

Demo¶

import org.apache.spark.sql.delta.commands.WriteIntoDelta
import org.apache.spark.sql.delta.DeltaLog
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.delta.DeltaOptions
val tableName = "/tmp/delta/t1"
val data = spark.range(5).toDF
val writeCmd = WriteIntoDelta(
  deltaLog = DeltaLog.forTable(spark, tableName),
  mode = SaveMode.Overwrite,
  options = new DeltaOptions(Map.empty[String, String], spark.sessionState.conf),
  partitionColumns = Seq.empty[String],
  configuration = Map.empty[String, String],
  data)

// Review web UI @ http://localhost:4040

writeCmd.run(spark)