spark-reviews mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cloud-fan <...@git.apache.org>
Subject [GitHub] spark pull request #19269: [SPARK-22026][SQL][WIP] data source v2 write path
Date Thu, 21 Sep 2017 08:11:58 GMT
Github user cloud-fan commented on a diff in the pull request:

    https://github.com/apache/spark/pull/19269#discussion_r140174897
  
    --- Diff: sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/DataSourceV2Writer.java
---
    @@ -0,0 +1,71 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one or more
    + * contributor license agreements.  See the NOTICE file distributed with
    + * this work for additional information regarding copyright ownership.
    + * The ASF licenses this file to You under the Apache License, Version 2.0
    + * (the "License"); you may not use this file except in compliance with
    + * the License.  You may obtain a copy of the License at
    + *
    + *    http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.spark.sql.sources.v2.writer;
    +
    +import org.apache.spark.annotation.InterfaceStability;
    +import org.apache.spark.sql.Row;
    +import org.apache.spark.sql.SaveMode;
    +import org.apache.spark.sql.sources.v2.DataSourceV2Options;
    +import org.apache.spark.sql.sources.v2.WriteSupport;
    +import org.apache.spark.sql.types.StructType;
    +
    +/**
    + * A data source writer that is returned by
    + * {@link WriteSupport#createWriter(StructType, SaveMode, DataSourceV2Options)}.
    + * It can mix in various writing optimization interfaces to speed up the data saving.
The actual
    + * writing logic is delegated to {@link WriteTask} that is returned by {@link #createWriteTask()}.
    + *
    + * The writing procedure is:
    + *   1. Create a write task by {@link #createWriteTask()}, serialize and send it to all
the
    + *      partitions of the input data(RDD).
    + *   2. For each partition, create a data writer with the write task, and write the data
of the
    + *      partition with this writer. If all the data are written successfully, call
    + *      {@link DataWriter#commit()}. If exception happens during the writing, call
    + *      {@link DataWriter#abort()}. This step may repeat several times as Spark will
retry failed
    + *      tasks.
    + *   3. Wait until all the writers/partitions are finished, i.e., either committed or
aborted. If
    + *      all partitions are written successfully, call {@link #commit(WriterCommitMessage[])}.
If
    + *      some partitions failed and aborted, call {@link #abort()}.
    + *
    + * Note that, data sources are responsible for providing transaction ability by implementing
the
    + * `commit` and `abort` methods of {@link DataSourceV2Writer} and {@link DataWriter}
correctly.
    + * The transaction here is Spark-level transaction, which may not be the underlying storage
    + * transaction. For example, Spark successfully write data to a Cassandra data source,
but
    + * Cassandra may need some more time to reach consistency at storage level.
    + */
    +@InterfaceStability.Evolving
    +public interface DataSourceV2Writer {
    +
    +  /**
    +   * Creates a write task which will be serialized and sent to executors. For each partition
of the
    +   * input data(RDD), there will be one write task to write the records.
    +   */
    +  WriteTask<Row> createWriteTask();
    +
    +  /**
    +   * Commits this writing job with a list of commit messages. The commit messages are
collected from
    +   * all data writers for this writing job and are produced by {@link DataWriter#commit()}.
This
    +   * also means all the data are written successfully and all data writers are committed.
    +   */
    +  void commit(WriterCommitMessage[] messages);
    +
    +  /**
    +   * Aborts this writing job because some data writers are failed to write the records
and aborted.
    +   */
    +  void abort();
    --- End diff --
    
    Let's say we have 400 partitions to write, and first 100 write tasks are successful and
committed. If the 101st write task failed, there is no way to re-launch the previous 100 write
tasks and abort them, we can only ask `DataSourceV2Writer` to do a job-level abort.
    
    So you are right, this method should also take a list of commit messages for already-successful
write tasks.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org


Mime
View raw message