apache / carbondata / 437

Build # 437

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/0896e8c9f7c9038eba04e0dbf76a965bd98db2fc">0896e8c9f">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/0896e8c9f7c9038eba04e0dbf76a965bd98db2fc">0896e8c9f</a><a href="https://github.com/apache/carbondata/commit/0896e8c9f7c9038eba04e0dbf76a965bd98db2fc">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/0896e8c9f7c9038eba04e0dbf76a965bd98db2fc&quot;&gt;0896e8c9f&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/0896e8c9f7c9038eba04e0dbf76a965bd98db2fc&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

173 of 173 new or added lines in 12 files covered. (100.0%)

58750 of 72884 relevant lines covered (80.61%)

1.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.5

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

/*
 * 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.carbondata.stream

import java.util.concurrent.{ConcurrentHashMap, CountDownLatch, TimeUnit}

import scala.collection.JavaConverters._

import org.apache.spark.sql.{CarbonEnv, DataFrame, SparkSession}
import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil
import org.apache.spark.sql.streaming.StreamingQuery
import org.apache.spark.sql.types.{StructField, StructType}

import org.apache.carbondata.common.exceptions.NoSuchStreamException
import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
import org.apache.carbondata.spark.StreamingOption
import org.apache.carbondata.streaming.CarbonStreamException

/**
 * A manager to start and stop a stream job for StreamSQL.
 * This stream job is only available to the driver memory and not persisted
 * so other drivers cannot see ongoing stream jobs.
 */
object StreamJobManager {
  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  // map of stream name to job desc
  private val jobs = new ConcurrentHashMap[String, StreamJobDesc]()

  private def validateSourceTable(source: CarbonTable): Unit = {
    if (!source.isStreamingSource) {
      throw new MalformedCarbonCommandException(
        s"Table ${source.getTableName} is not streaming source table " +
        "('streaming' tblproperty is not 'source')")
    }
  }

  private def validateSinkTable(validateQuerySchema: Boolean,
                                querySchema: StructType, sink: CarbonTable): Unit = {
    if (!sink.isStreamingSink) {
      throw new MalformedCarbonCommandException(
        s"Table ${sink.getTableName} is not streaming sink table " +
        "('streaming' tblproperty is not 'sink' or 'true')")
    }
    if (validateQuerySchema) {
      val fields = sink.getCreateOrderColumn(sink.getTableName).asScala.map { column =>
        StructField(
          column.getColName,
          CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(column.getDataType))
      }
      if (!querySchema.equals(StructType(fields))) {
        throw new MalformedCarbonCommandException(
          s"Schema of table ${sink.getTableName} does not match query output")
      }
    }
  }

  /**
   * Start a spark streaming job
   * @param sparkSession session instance
   * @param ifNotExists if not exists is set or not
   * @param streamName name of the stream
   * @param sourceTable stream source table
   * @param sinkTable sink table to insert to
   * @param query query string
   * @param streamDf dataframe that containing the query from stream source table
   * @param options options provided by user
   * @return Job ID
   */
  def startStream(
      sparkSession: SparkSession,
      ifNotExists: Boolean,
      streamName: String,
      sourceTable: CarbonTable,
      sinkTable: CarbonTable,
      query: String,
      streamDf: DataFrame,
      options: StreamingOption): String = {
    val latch = new CountDownLatch(1)
    var exception: Throwable = null
    var job: StreamingQuery = null

    if (jobs.containsKey(streamName)) {
      if (ifNotExists) {
        return jobs.get(streamName).streamingQuery.id.toString
      } else {
        throw new MalformedCarbonCommandException(s"Stream Name $streamName already exists")
      }
    }

    validateSourceTable(sourceTable)

    // for kafka and socket stream source, the source table schema is one string column only
    // so we do not validate the query schema against the sink table schema
    val isKafka = Option(sourceTable.getFormat).contains("kafka")
    val isSocket = Option(sourceTable.getFormat).contains("socket")
    validateSinkTable(!(isKafka || isSocket), streamDf.schema, sinkTable)

    // start a new thread to run the streaming ingest job, the job will be running
    // until user stops it by STOP STREAM JOB
    val thread = new Thread(new Runnable {
      override def run(): Unit = {
        try {
          job = streamDf.writeStream
            .format("carbondata")
            .trigger(options.trigger)
            .option("checkpointLocation", options.checkpointLocation(sinkTable.getTablePath))
            .option("dateformat", options.dateFormat)
            .option("timestampformat", options.timeStampFormat)
            .option("carbon.stream.parser", options.rowParser)
            .option("dbName", sinkTable.getDatabaseName)
            .option("tableName", sinkTable.getTableName)
            .option("bad_record_path", options.badRecordsPath)
            .option("bad_records_action", options.badRecordsAction)
            .option("bad_records_logger_enable", options.badRecordsLogger)
            .option("is_empty_bad_record", options.isEmptyBadRecord)
            .options(options.remainingOption)
            .start()
          latch.countDown()
          job.awaitTermination()
        } catch {
          case e: Throwable =>
            LOGGER.error(e)
            exception = e
            latch.countDown()
        }
      }
    })
    thread.start()

    // wait for max 10 seconds for the streaming job to start
    if (latch.await(10, TimeUnit.SECONDS)) {
      if (exception != null) {
        throw exception
      }

      jobs.put(
        streamName,
        StreamJobDesc(job, streamName, sourceTable.getDatabaseName, sourceTable.getTableName,
          sinkTable.getDatabaseName, sinkTable.getTableName, query, thread))

      LOGGER.audit(s"STREAM $streamName started with job id '${job.id.toString}', " +
                   s"from ${sourceTable.getDatabaseName}.${sourceTable.getTableName} " +
                   s"to ${sinkTable.getDatabaseName}.${sinkTable.getTableName}")
      job.id.toString
    } else {
      thread.interrupt()
      throw new CarbonStreamException("Streaming job takes too long to start")
    }
  }

  /**
   * Stop a streaming job
   * @param streamName name of the stream
   * @param ifExists if exists is set or not
   */
  def stopStream(streamName: String, ifExists: Boolean): Unit = {
    if (jobs.containsKey(streamName)) {
      val jobDesc = jobs.get(streamName)
      jobDesc.streamingQuery.stop()
      jobDesc.thread.interrupt()
      jobs.remove(streamName)
      LOGGER.audit(s"STREAM $streamName stopped, job id '${jobDesc.streamingQuery.id.toString}', " +
                   s"from ${jobDesc.sourceDb}.${jobDesc.sourceTable} " +
                   s"to ${jobDesc.sinkDb}.${jobDesc.sinkTable}")
    } else {
      if (!ifExists) {
        throw new NoSuchStreamException(streamName)
      }
    }
  }

  /**
   * Return all running jobs
   * @return running jobs
   */
  def getAllJobs: Set[StreamJobDesc] = jobs.values.asScala.toSet

}

/**
 * A job description for the StreamSQL job
 */
private[stream] case class StreamJobDesc(
    streamingQuery: StreamingQuery,
    streamName: String,
    sourceDb: String,
    sourceTable: String,
    sinkDb: String,
    sinkTable: String,
    query: String,
    thread: Thread,
    startTime: Long = System.currentTimeMillis()
)

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.stream
19
20	import java.util.concurrent.{ConcurrentHashMap, CountDownLatch, TimeUnit}
21
22	import scala.collection.JavaConverters._
23
24	import org.apache.spark.sql.{CarbonEnv, DataFrame, SparkSession}
25	import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil
26	import org.apache.spark.sql.streaming.StreamingQuery
27	import org.apache.spark.sql.types.{StructField, StructType}
28
29	import org.apache.carbondata.common.exceptions.NoSuchStreamException
30	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
31	import org.apache.carbondata.common.logging.LogServiceFactory
32	import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
33	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
34	import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
35	import org.apache.carbondata.spark.StreamingOption
36	import org.apache.carbondata.streaming.CarbonStreamException
37
38	/**
39	* A manager to start and stop a stream job for StreamSQL.
40	* This stream job is only available to the driver memory and not persisted
41	* so other drivers cannot see ongoing stream jobs.
42	*/
43	object StreamJobManager {
44	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	2✔
45
46	// map of stream name to job desc
47	private val jobs = new ConcurrentHashMap[String, StreamJobDesc]()	2✔
48
49	private def validateSourceTable(source: CarbonTable): Unit = {
50	if (!source.isStreamingSource) {	2✔
51	throw new MalformedCarbonCommandException(	×
52	s"Table ${source.getTableName} is not streaming source table " +	×
53	"('streaming' tblproperty is not 'source')")	×
54	}
55	}
56
57	private def validateSinkTable(validateQuerySchema: Boolean,
58	querySchema: StructType, sink: CarbonTable): Unit = {	2✔
59	if (!sink.isStreamingSink) {	2✔
60	throw new MalformedCarbonCommandException(	×
61	s"Table ${sink.getTableName} is not streaming sink table " +	×
62	"('streaming' tblproperty is not 'sink' or 'true')")	×
63	}
64	if (validateQuerySchema) {	2✔
65	val fields = sink.getCreateOrderColumn(sink.getTableName).asScala.map { column =>	2✔
66	StructField(	2✔
67	column.getColName,	2✔
68	CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(column.getDataType))	2✔
69	}
70	if (!querySchema.equals(StructType(fields))) {	2✔
71	throw new MalformedCarbonCommandException(	×
72	s"Schema of table ${sink.getTableName} does not match query output")	×
73	}
74	}
75	}
76
77	/**
78	* Start a spark streaming job
79	* @param sparkSession session instance
80	* @param ifNotExists if not exists is set or not
81	* @param streamName name of the stream
82	* @param sourceTable stream source table
83	* @param sinkTable sink table to insert to
84	* @param query query string
85	* @param streamDf dataframe that containing the query from stream source table
86	* @param options options provided by user
87	* @return Job ID
88	*/
89	def startStream(
90	sparkSession: SparkSession,
91	ifNotExists: Boolean,
92	streamName: String,
93	sourceTable: CarbonTable,
94	sinkTable: CarbonTable,
95	query: String,
96	streamDf: DataFrame,
97	options: StreamingOption): String = {	2✔
98	val latch = new CountDownLatch(1)	2✔
99	var exception: Throwable = null	2✔
100	var job: StreamingQuery = null	2✔
101
102	if (jobs.containsKey(streamName)) {	2✔
103	if (ifNotExists) {	2✔
104	return jobs.get(streamName).streamingQuery.id.toString	2✔
105	} else {
106	throw new MalformedCarbonCommandException(s"Stream Name $streamName already exists")	×
107	}
108	}
109
110	validateSourceTable(sourceTable)	2✔
111
112	// for kafka and socket stream source, the source table schema is one string column only
113	// so we do not validate the query schema against the sink table schema
114	val isKafka = Option(sourceTable.getFormat).contains("kafka")	2✔
115	val isSocket = Option(sourceTable.getFormat).contains("socket")	2✔
116	validateSinkTable(!(isKafka \|\| isSocket), streamDf.schema, sinkTable)	2✔
117
118	// start a new thread to run the streaming ingest job, the job will be running
119	// until user stops it by STOP STREAM JOB
120	val thread = new Thread(new Runnable {	2✔
121	override def run(): Unit = {
122	try {	2✔
123	job = streamDf.writeStream	2✔
124	.format("carbondata")	2✔
125	.trigger(options.trigger)	2✔
126	.option("checkpointLocation", options.checkpointLocation(sinkTable.getTablePath))	2✔
127	.option("dateformat", options.dateFormat)	2✔
128	.option("timestampformat", options.timeStampFormat)	2✔
129	.option("carbon.stream.parser", options.rowParser)	2✔
130	.option("dbName", sinkTable.getDatabaseName)	2✔
131	.option("tableName", sinkTable.getTableName)	2✔
132	.option("bad_record_path", options.badRecordsPath)	2✔
133	.option("bad_records_action", options.badRecordsAction)	2✔
134	.option("bad_records_logger_enable", options.badRecordsLogger)	2✔
135	.option("is_empty_bad_record", options.isEmptyBadRecord)	2✔
136	.options(options.remainingOption)	2✔
137	.start()	2✔
138	latch.countDown()	2✔
139	job.awaitTermination()	2✔
140	} catch {
141	case e: Throwable =>
142	LOGGER.error(e)	2✔
143	exception = e	2✔
144	latch.countDown()	2✔
145	}
146	}
147	})
148	thread.start()	2✔
149
150	// wait for max 10 seconds for the streaming job to start
151	if (latch.await(10, TimeUnit.SECONDS)) {	2✔
152	if (exception != null) {	2✔
153	throw exception	2✔
154	}
155
156	jobs.put(	2✔
157	streamName,	2✔
158	StreamJobDesc(job, streamName, sourceTable.getDatabaseName, sourceTable.getTableName,	2✔
159	sinkTable.getDatabaseName, sinkTable.getTableName, query, thread))	2✔
160
161	LOGGER.audit(s"STREAM $streamName started with job id '${job.id.toString}', " +	2✔
162	s"from ${sourceTable.getDatabaseName}.${sourceTable.getTableName} " +	2✔
163	s"to ${sinkTable.getDatabaseName}.${sinkTable.getTableName}")	2✔
164	job.id.toString	2✔
165	} else {
166	thread.interrupt()	×
167	throw new CarbonStreamException("Streaming job takes too long to start")	×
168	}
169	}
170
171	/**
172	* Stop a streaming job
173	* @param streamName name of the stream
174	* @param ifExists if exists is set or not
175	*/
176	def stopStream(streamName: String, ifExists: Boolean): Unit = {
177	if (jobs.containsKey(streamName)) {	2✔
178	val jobDesc = jobs.get(streamName)	2✔
179	jobDesc.streamingQuery.stop()	2✔
180	jobDesc.thread.interrupt()	2✔
181	jobs.remove(streamName)	2✔
182	LOGGER.audit(s"STREAM $streamName stopped, job id '${jobDesc.streamingQuery.id.toString}', " +	2✔
183	s"from ${jobDesc.sourceDb}.${jobDesc.sourceTable} " +	2✔
184	s"to ${jobDesc.sinkDb}.${jobDesc.sinkTable}")	2✔
185	} else {
186	if (!ifExists) {	2✔
187	throw new NoSuchStreamException(streamName)	2✔
188	}
189	}
190	}
191
192	/**
193	* Return all running jobs
194	* @return running jobs
195	*/
196	def getAllJobs: Set[StreamJobDesc] = jobs.values.asScala.toSet	2✔
197
198	}
199
200	/**
201	* A job description for the StreamSQL job
202	*/
203	private[stream] case class StreamJobDesc(	2✔
204	streamingQuery: StreamingQuery,	2✔
205	streamName: String,	2✔
206	sourceDb: String,	2✔
207	sourceTable: String,	2✔
208	sinkDb: String,	2✔
209	sinkTable: String,	2✔
210	query: String,	2✔
211	thread: Thread,	2✔
212	startTime: Long = System.currentTimeMillis()	2✔
213	)

apache / carbondata / 437

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous