Spark Data Source (finally) #266

vinothchandar · 2017-09-27T18:17:40Z

Fixes #7

vinothchandar · 2017-09-27T18:18:09Z

I am going to work on merging the commits into 1 in the meantime.

zqureshi

Minor comments, need to finish last third.

zqureshi · 2017-09-29T02:51:37Z

docs/configurations.md

+    <span style="color:grey"> Options useful for writing datasets </span>
+        - [OPERATION_OPT_KEY](#OPERATION_OPT_KEY) (Default: upsert) <br/>
+        <span style="color:grey">whether to do upsert, insert or bulkinsert for the write operation</span>
+        - [STORAGE_TYPE_OPT_KEY](#STORAGE_TYPE_OPT_KEY) (Default: COPY_ON_WRITE) <br/>


Could this change if you're updating an already existing table?

Technically, you can write to a COW table as MOR going forward. But yes, table name and storage type can't change across commits..

added a note to that effect.

zqureshi · 2017-09-29T02:57:51Z

docs/incremental_processing.md

+             DataSourceReadOptions.VIEW_TYPE_INCREMENTAL_OPT_VAL())
+     .option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY(),
+            <beginInstantTime>)
+     .load(tablePath); // For incremental view, pass in the root/base path of dataset


For non-incremental view is it possible to pass in anything other than base path of hoodie table to read?

yes. you can pass in glob with a mix of multiple datasets with both hoodie and non-hoodie ones for the RO view.

zqureshi · 2017-09-29T03:11:30Z

hoodie-client/src/test/java/com/uber/hoodie/TestHoodieClientOnCopyOnWriteStorage.java

        HoodieTableMetaClient metaClient = new HoodieTableMetaClient(fs, basePath);
+        HoodieTimeline timeline = new HoodieActiveTimeline(fs, metaClient.getMetaPath()).getCommitTimeline();


Why the switch to testing this via timeline vs client? I see you're using HoodieClientTestUtils below for reading.

since I moved all the incremental pulling & commit listing methods etc from HoodieReadClient, had to use the timeline directly to have these tests going again.. Hope that makes sense.

zqureshi · 2017-09-29T04:06:13Z

hoodie-client/src/test/java/com/uber/hoodie/common/HoodieClientTestUtils.java

@@ -90,4 +106,83 @@ public static SparkConf getSparkConfForTest(String appName) {
                .setMaster("local[1]");
        return HoodieReadClient.addHoodieSupport(sparkConf);
    }
+
+    public static HashMap<String, String> getLatestFileIsToFullPath(String basePath,


s/FileIs/FileId/

oops. will fix.

zqureshi · 2017-09-29T04:23:31Z

hoodie-spark/src/main/java/com/uber/hoodie/DataSourceUtils.java

+            if (i == parts.length - 1) {
+                return val.toString();
+            } else {
+                if (val instanceof GenericRecord) {


Is this supposed to be not (val instanceof GenericRecord)?

yes. I have been testing with top level names all along. Will try to add a test around this as well

zqureshi · 2017-09-29T04:25:25Z

hoodie-spark/src/main/java/com/uber/hoodie/DataSourceUtils.java

+    }
+
+    /**
+     * Create a payload class via reflection, passing in an ordering/precombine value value.


s/value value/value/

Same. will do

zqureshi · 2017-09-29T04:40:10Z

hoodie-spark/src/main/scala/com/uber/hoodie/AvroConversionUtils.scala

+            targetList
+          }
+        }
+      case MapType(StringType, valueType, _) =>


Note: Reviewed till here, continuing tomorrow.

thanks @zqureshi . I know this is a big one :)

Lwelch45

My comments are around style mostly.

Lwelch45 · 2017-09-29T23:39:09Z

hoodie-spark/src/main/java/com/uber/hoodie/HoodieDataSourceHelpers.java

+     * Get a list of instant times that have occurred, from the given instant timestamp.
+     *
+     * @param instantTimestamp
+     * @return


This is a nitpick but I'd remove the empty @return's in the function docs

Removed what I saw. Fishing this out and enforcing across teh board is out of scope of this PR :)

Lwelch45 · 2017-09-30T00:09:00Z

hoodie-utilities/src/main/java/com/uber/hoodie/utilities/deltastreamer/HoodieDeltaStreamer.java

@@ -349,7 +348,6 @@ public SourceDataFormat convert(String value) throws ParameterException {
    public static void main(String[] args) throws Exception {
        final Config cfg = new Config();
        JCommander cmd = new JCommander(cfg, args);
-        // TODO(vc): Do proper validation
        if (cfg.help || args.length == 0) {


maybe include a check of to see if none of the other args have been matched.

Lwelch45 · 2017-09-30T00:11:26Z

hoodie-spark/src/test/scala/DataSourceTest.scala

+        .load(basePath + "/*/*/*/*")
+      fail() // we would error out, since no compaction has yet occurred.
+    } catch {
+      case e: Exception => {


Assert the exception is the type that you expect

Lwelch45 · 2017-09-30T00:12:24Z

hoodie-spark/src/test/scala/DataSourceTest.scala

+    try {
+      val hoodieROViewDF1 = spark.read.format("com.uber.hoodie")
+        .load(basePath + "/*/*/*/*")
+      fail() // we would error out, since no compaction has yet occurred.


add a message to the failure

Lwelch45 · 2017-09-30T00:13:39Z

hoodie-spark/src/test/java/HoodieJavaApp.java

+        cli.run();
+    }
+
+


nitpick 1 newline between method calls

prazanna

Looks Good. Lets gets this baby merged.

prazanna · 2017-10-02T03:27:17Z

hoodie-spark/src/test/java/HoodieJavaApp.java

+        Dataset<Row> hoodieROViewDF = spark.read()
+                .format("com.uber.hoodie")
+                // pass any path glob, can include hoodie & non-hoodie datasets
+                .load(tablePath + "/*/*/*/*");


Why pass a pattern to glob? Do we not know this from partition metadata?

Because spark.read.parquet takes a glob of paths. Idea is to support a mix of hoodie/non-hoodie paths.. Underneath, we do look up partition metadata to check if this is a hoodie dataset and so forth.

prazanna · 2017-10-02T03:28:35Z

hoodie-spark/src/test/java/HoodieJavaApp.java

+                .format("com.uber.hoodie")
+                // pass any path glob, can include hoodie & non-hoodie datasets
+                .load(tablePath + "/*/*/*/*");
+        hoodieROViewDF.registerTempTable("hoodie_ro");


Can we pass a SQL statement during the read and register it as a temp table? It will be good if we can do that.

we can. a sql statement is executed on this table in lines below.

- Write with COW/MOR paths work fully - Read with RO view works on both storages* - Incremental view supported on COW - Refactored out HoodieReadClient methods, to just contain key based access - HoodieDataSourceHelpers class can be now used to construct inputs to datasource - Tests in hoodie-client using new helpers and mechanisms - Basic tests around save modes & insert/upserts (more to follow) - Bumped up scala to 2.11, since 2.10 is deprecated & complains with scalatest - Updated documentation to describe usage - New sample app written using the DataSource API

zqureshi · 2017-10-03T02:04:13Z

hoodie-spark/src/main/scala/com/uber/hoodie/DataSourceOptions.scala

+    * Default: upsert()
+    */
+  val OPERATION_OPT_KEY = "hoodie.datasource.write.operation"
+  val BULK_INSERT_OPERATION_OPT_VAL = "bulk_insert"


All read constant values are upper case whereas all write are lower. Is that intentional?

Not intentional. But write constant are named consistently as below right

val DEFAULT_PRECOMBINE_FIELD_OPT_VAL = "ts"

all vals are upper case .. May be I am not following along

val VIEW_TYPE_INCREMENTAL_OPT_VAL = "INCREMENTAL"
vs
val BULK_INSERT_OPERATION_OPT_VAL = "bulk_insert"

I over generalized, but basically my question was that the value is all caps in the case of some constants but most others are lower.

Ah okay. no reason. made it consistent now.

zqureshi · 2017-10-03T02:28:22Z

hoodie-spark/src/main/scala/com/uber/hoodie/IncrementalRelation.scala

+  // use schema from a file produced in the latest instant
+  val latestSchema = {
+    val latestMeta = HoodieCommitMetadata
+      .fromBytes(commitTimeline.getInstantDetails(commitsToReturn.last).get)


This is quite neat. :)

:) as long as things are backwards compatible. queries should work!

zqureshi

Looking great! 🚢

vinothchandar · 2017-10-03T02:45:14Z

Thanks @zqureshi for the diligent review.. Addressed all comments and added a few tests too. Merging soon ..

vinothchandar self-assigned this Sep 27, 2017

vinothchandar requested review from zqureshi and prazanna September 27, 2017 18:17

vinothchandar force-pushed the datasource branch from 70ab733 to c41b0c7 Compare September 27, 2017 18:23

zqureshi reviewed Sep 29, 2017

View reviewed changes

Lwelch45 reviewed Sep 30, 2017

View reviewed changes

prazanna approved these changes Oct 2, 2017

View reviewed changes

zqureshi reviewed Oct 3, 2017

View reviewed changes

zqureshi approved these changes Oct 3, 2017

View reviewed changes

Incorporating code review feedback for DataSource

31cd0e9

vinothchandar force-pushed the datasource branch from c41b0c7 to 31cd0e9 Compare October 3, 2017 03:32

vinothchandar merged commit 274aaf4 into apache:master Oct 3, 2017

		HoodieTableMetaClient metaClient = new HoodieTableMetaClient(fs, basePath);
		HoodieTimeline timeline = new HoodieActiveTimeline(fs, metaClient.getMetaPath()).getCommitTimeline();

Spark Data Source (finally) #266

Spark Data Source (finally) #266

Conversation

vinothchandar commented Sep 27, 2017

vinothchandar commented Sep 27, 2017

zqureshi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Lwelch45 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

prazanna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zqureshi left a comment

Choose a reason for hiding this comment

vinothchandar commented Oct 3, 2017