[CARBONDATA-2919] Support ingest from Kafka in StreamSQL #2695

jackylk · 2018-09-05T17:47:03Z

In this PR, following is changed:

Support Kafka as streaming source table in StreamSQL
Delete the lock file when DROP STREAM is triggered
Fix the streaming segment data size shown in SHOW SEGMENTS

Any interfaces changed?
NA
Any backward compatibility impacted?
In SHOW SEGMENT command, for Load Start Time and End Time, the output data type in RDD is changed from timestamp to string.
Document update required?
NA
Testing done
Please provide details on
- Whether new unit test cases have been added or why no new tests are required?
- How it is tested? Please attach test report.
- Is it a performance related change? Please attach the performance test report.
- Any additional information to help reviewers in testing this change.
rerun all test
For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
NA

CarbonDataQA · 2018-09-05T18:30:09Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/100/

CarbonDataQA · 2018-09-05T19:05:08Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/268/

CarbonDataQA · 2018-09-05T19:57:05Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8338/

QiangCai · 2018-09-06T03:11:52Z

examples/spark2/pom.xml

+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql-kafka-0-10_${scala.binary.version}</artifactId>
+      <version>2.2.1</version>


please use spark.version property

QiangCai · 2018-09-06T03:13:29Z

examples/spark2/src/main/scala/org/apache/carbondata/examples/StreamSQLExample.scala

+    System.out.println("streaming finished")
+  }
+
+  def showTableCount(spark: SparkSession, tableName: String): Thread = {


please remove unused code

CarbonDataQA · 2018-09-06T07:10:53Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/111/

CarbonDataQA · 2018-09-06T07:42:28Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/279/

CarbonDataQA · 2018-09-06T07:46:18Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8349/

xuchuanyin · 2018-09-07T14:08:58Z

core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java

+   */
+  public String getFormat() {
+    return getTableInfo().getFactTable().getTableProperties()
+            .get("format");


it seems this can be moved to the previous line

xuchuanyin · 2018-09-07T14:10:48Z

integration/spark2/src/main/scala/org/apache/carbondata/stream/StreamJobManager.scala

-    validateSinkTable(streamDf.schema, sinkTable)
+
+    // kafka source always have fixed schema, need to get actual schema
+    val isKafka = Option(sourceTable.getFormat).exists(_ == "kafka")


for string equality, better to use equals

xuchuanyin · 2018-09-07T14:13:50Z

integration/spark2/src/main/scala/org/apache/carbondata/stream/StreamJobManager.scala

+          CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(column.getDataType))
+      }
+      if (!querySchema.equals(StructType(fields))) {
+        throw new MalformedCarbonCommandException(s"Schema of table ${ sink.getTableName } " +


you can move the first half to the next line so that we can avoid string concatenation here.
The same with line#58

xuchuanyin · 2018-09-07T14:15:28Z

...src/main/scala/org/apache/spark/sql/execution/command/stream/CarbonCreateStreamCommand.scala

-    val streamReader = sparkSession.readStream
-      .schema(getSparkSchema(sourceTable))
-      .format(format)
+    val streamReader = if (format != "kafka") {


use 'equals' instead of '='

xuchuanyin · 2018-09-07T14:16:44Z

streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java

@@ -316,6 +316,10 @@ private void appendBlockletToDataFile() throws IOException {
  }

  public BlockletMinMaxIndex getBatchMinMaxIndex() {
+    if (output == null) {
+      return StreamSegment.mergeBlockletMinMax(


no need for multiple lines here

xuchuanyin · 2018-09-07T14:18:19Z

pom.xml

@@ -217,6 +217,12 @@
        <version>${spark.version}</version>
        <scope>${spark.deps.scope}</scope>
      </dependency>
+      <dependency>


Why is it needed for whole project scope? I think it will only be used in streaming related modules

CarbonDataQA · 2018-09-07T17:27:25Z

Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/162/

CarbonDataQA · 2018-09-07T18:27:35Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/330/

CarbonDataQA · 2018-09-07T18:29:52Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8401/

chenliang613 · 2018-09-08T02:20:51Z

retest this please

CarbonDataQA · 2018-09-08T02:32:29Z

Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/164/

CarbonDataQA · 2018-09-08T03:32:56Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/332/

CarbonDataQA · 2018-09-08T03:35:25Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8403/

CarbonDataQA · 2018-09-10T10:15:29Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8435/

CarbonDataQA · 2018-09-10T10:16:07Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/197/

CarbonDataQA · 2018-09-10T11:22:48Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/365/

CarbonDataQA · 2018-09-10T13:02:17Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/371/

CarbonDataQA · 2018-09-10T13:02:20Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8441/

CarbonDataQA · 2018-09-10T13:05:21Z

Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/203/

CarbonDataQA · 2018-09-10T13:07:15Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8442/

CarbonDataQA · 2018-09-10T13:07:16Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/372/

jackylk · 2018-09-11T01:08:31Z

retest this please

CarbonDataQA · 2018-09-11T01:10:18Z

Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/214/

CarbonDataQA · 2018-09-11T01:12:17Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/383/

CarbonDataQA · 2018-09-11T01:12:17Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8453/

CarbonDataQA · 2018-09-11T16:49:41Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/238/

CarbonDataQA · 2018-09-11T17:50:18Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/407/

CarbonDataQA · 2018-09-12T03:49:52Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/241/

CarbonDataQA · 2018-09-12T03:50:42Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8480/

CarbonDataQA · 2018-09-12T04:56:52Z

Build Success with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/410/

brijoobopanna · 2018-09-12T08:17:58Z

retest this please

CarbonDataQA · 2018-09-12T08:29:51Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/252/

CarbonDataQA · 2018-09-12T08:30:48Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8491/

CarbonDataQA · 2018-09-12T09:59:21Z

Build Success with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/421/

CarbonDataQA · 2018-09-12T10:49:31Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8492/

CarbonDataQA · 2018-09-12T10:49:47Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/253/

CarbonDataQA · 2018-09-12T11:39:23Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/256/

CarbonDataQA · 2018-09-12T11:57:40Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8495/

CarbonDataQA · 2018-09-12T11:58:06Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/422/

CarbonDataQA · 2018-09-12T13:11:43Z

Build Success with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/425/

fix compile fix comment fix test fix spark2.3

CarbonDataQA · 2018-09-13T01:50:25Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/263/

CarbonDataQA · 2018-09-13T02:11:07Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8502/

CarbonDataQA · 2018-09-13T03:19:38Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/267/

CarbonDataQA · 2018-09-13T03:21:27Z

Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/432/

CarbonDataQA · 2018-09-13T03:40:52Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8506/

CarbonDataQA · 2018-09-13T04:27:50Z

Build Success with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/436/

jackylk · 2018-09-13T06:26:23Z

retest this please

CarbonDataQA · 2018-09-13T06:39:32Z

Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/268/

CarbonDataQA · 2018-09-13T06:53:37Z

Build Failed with Spark 2.3.1, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.3/8507/

CarbonDataQA · 2018-09-13T07:43:12Z

Build Success with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/437/

QiangCai reviewed Sep 6, 2018

View reviewed changes

xuchuanyin reviewed Sep 7, 2018

View reviewed changes

jackylk force-pushed the kafka branch from 8f3f58f to 393347e Compare September 7, 2018 17:16

jackylk force-pushed the kafka branch from 7da3e6f to 4deb99a Compare September 10, 2018 13:00

jackylk force-pushed the kafka branch from 4deb99a to 0ed75bb Compare September 10, 2018 13:02

jackylk force-pushed the kafka branch from 683a624 to 16f8987 Compare September 12, 2018 10:38

support kafka in StreamSQL

b146695

fix compile fix comment fix test fix spark2.3

jackylk force-pushed the kafka branch from 1a409ec to b146695 Compare September 13, 2018 01:38

fix compile

eefe0c2

asfgit closed this in 560bfbe Sep 13, 2018

[CARBONDATA-2919] Support ingest from Kafka in StreamSQL #2695

[CARBONDATA-2919] Support ingest from Kafka in StreamSQL #2695

Conversation

jackylk commented Sep 5, 2018 • edited

CarbonDataQA commented Sep 5, 2018

CarbonDataQA commented Sep 5, 2018

CarbonDataQA commented Sep 5, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

CarbonDataQA commented Sep 6, 2018

CarbonDataQA commented Sep 6, 2018

CarbonDataQA commented Sep 6, 2018

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

CarbonDataQA commented Sep 7, 2018

CarbonDataQA commented Sep 7, 2018

CarbonDataQA commented Sep 7, 2018

chenliang613 commented Sep 8, 2018

CarbonDataQA commented Sep 8, 2018

CarbonDataQA commented Sep 8, 2018

CarbonDataQA commented Sep 8, 2018

CarbonDataQA commented Sep 10, 2018

CarbonDataQA commented Sep 10, 2018

CarbonDataQA commented Sep 10, 2018

CarbonDataQA commented Sep 10, 2018

CarbonDataQA commented Sep 10, 2018

CarbonDataQA commented Sep 10, 2018

CarbonDataQA commented Sep 10, 2018

CarbonDataQA commented Sep 10, 2018

jackylk commented Sep 11, 2018

CarbonDataQA commented Sep 11, 2018

CarbonDataQA commented Sep 11, 2018

CarbonDataQA commented Sep 11, 2018

CarbonDataQA commented Sep 11, 2018

CarbonDataQA commented Sep 11, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

brijoobopanna commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 12, 2018

CarbonDataQA commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

jackylk commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

CarbonDataQA commented Sep 13, 2018

jackylk commented Sep 5, 2018 •

edited