You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ma...@apache.org on 2018/01/19 03:53:58 UTC
carbondata git commit: [CARBONDATA-2053] Added events for streaming
Repository: carbondata
Updated Branches:
refs/heads/master 543a903c6 -> 2d9de5029
[CARBONDATA-2053] Added events for streaming
This PR contains the code to fire events during following two steps :
1. Before starting the stream and after the completion of stream
2. Before adding a batch and after batch is added during stream.
This closes #1832
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/2d9de502
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/2d9de502
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/2d9de502
Branch: refs/heads/master
Commit: 2d9de502951d124e4b87fe6d38434f5dd05ae7c2
Parents: 543a903
Author: rahulforallp <ra...@knoldus.in>
Authored: Thu Jan 18 17:24:24 2018 +0530
Committer: manishgupta88 <to...@gmail.com>
Committed: Fri Jan 19 09:26:07 2018 +0530
----------------------------------------------------------------------
.../streaming/StreamSinkFactory.scala | 33 +++++++++++++++++---
.../streaming/CarbonAppendableStreamSink.scala | 25 +++++++++++++--
2 files changed, 50 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/2d9de502/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala b/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
index cecc18c..1c7be6a 100644
--- a/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
+++ b/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
@@ -32,7 +32,8 @@ import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonStorePath
-import org.apache.carbondata.hadoop.streaming.CarbonStreamOutputFormat
+import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
+import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePostExecutionEvent, LoadTablePreExecutionEvent}
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.spark.dictionary.provider.SecureDictionaryServiceProvider
import org.apache.carbondata.spark.dictionary.server.SecureDictionaryServer
@@ -51,15 +52,29 @@ object StreamSinkFactory {
parameters: Map[String, String]): Sink = {
validateParameters(parameters)
- // prepare the stream segment
- val segmentId = getStreamSegmentId(carbonTable)
// build load model
val carbonLoadModel = buildCarbonLoadModelForStream(
sparkSession,
hadoopConf,
carbonTable,
parameters,
- segmentId)
+ "")
+ // fire pre event before streamin is started
+ val operationContext = new OperationContext
+ val loadTablePreExecutionEvent = new LoadTablePreExecutionEvent(
+ carbonTable.getCarbonTableIdentifier,
+ carbonLoadModel,
+ carbonLoadModel.getFactFilePath,
+ false,
+ parameters.asJava,
+ null,
+ false
+ )
+ OperationListenerBus.getInstance().fireEvent(loadTablePreExecutionEvent, operationContext)
+ // prepare the stream segment
+ val segmentId = getStreamSegmentId(carbonTable)
+ carbonLoadModel.setSegmentId(segmentId)
+
// start server if necessary
val server = startDictionaryServer(
sparkSession,
@@ -71,13 +86,21 @@ object StreamSinkFactory {
carbonLoadModel.setUseOnePass(false)
}
// default is carbon appended stream sink
- new CarbonAppendableStreamSink(
+ val carbonAppendableStreamSink = new CarbonAppendableStreamSink(
sparkSession,
carbonTable,
segmentId,
parameters,
carbonLoadModel,
server)
+
+ // fire post event before streamin is started
+ val loadTablePostExecutionEvent = new LoadTablePostExecutionEvent(
+ carbonTable.getCarbonTableIdentifier,
+ carbonLoadModel
+ )
+ OperationListenerBus.getInstance().fireEvent(loadTablePostExecutionEvent, operationContext)
+ carbonAppendableStreamSink
}
private def validateParameters(parameters: Map[String, String]): Unit = {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/2d9de502/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala b/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
index ce9446f..849bf99 100644
--- a/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
+++ b/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
@@ -19,13 +19,12 @@ package org.apache.spark.sql.execution.streaming
import java.util.Date
+import scala.collection.JavaConverters._
+
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce._
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
-import org.apache.hadoop.mapreduce.v2.api.records.JobId
-import org.apache.spark._
import org.apache.spark.TaskContext
-import org.apache.spark.internal.io._
import org.apache.spark.internal.io.FileCommitProtocol
import org.apache.spark.internal.io.FileCommitProtocol.TaskCommitMessage
import org.apache.spark.sql.{DataFrame, SparkSession}
@@ -42,8 +41,10 @@ import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.stats.QueryStatistic
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonStorePath
+import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
import org.apache.carbondata.hadoop.streaming.CarbonStreamOutputFormat
import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
+import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePostExecutionEvent, LoadTablePreExecutionEvent}
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.streaming.{CarbonStreamException, StreamHandoffRDD}
import org.apache.carbondata.streaming.parser.CarbonStreamParser
@@ -93,6 +94,18 @@ class CarbonAppendableStreamSink(
val statistic = new QueryStatistic()
+ // fire pre event on every batch add
+ val operationContext = new OperationContext
+ val loadTablePreExecutionEvent = new LoadTablePreExecutionEvent(
+ carbonTable.getCarbonTableIdentifier,
+ carbonLoadModel,
+ carbonLoadModel.getFactFilePath,
+ false,
+ parameters.asJava,
+ null,
+ false
+ )
+ OperationListenerBus.getInstance().fireEvent(loadTablePreExecutionEvent, operationContext)
checkOrHandOffSegment()
// committer will record how this spark job commit its output
@@ -119,6 +132,12 @@ class CarbonAppendableStreamSink(
hadoopConf,
carbonLoadModel,
server)
+ // fire post event on every batch add
+ val loadTablePostExecutionEvent = new LoadTablePostExecutionEvent(
+ carbonTable.getCarbonTableIdentifier,
+ carbonLoadModel
+ )
+ OperationListenerBus.getInstance().fireEvent(loadTablePostExecutionEvent, operationContext)
statistic.addStatistics(s"add batch: $batchId", System.currentTimeMillis())
CarbonAppendableStreamSink.LOGGER.info(