HeartSaVioR commented on a change in pull request #34333: URL: https://github.com/apache/spark/pull/34333#discussion_r739943400
########## File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/RatePerMicroBatchProvider.scala ########## @@ -0,0 +1,127 @@ +/* + * 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.execution.streaming.sources + +import java.util + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability} +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder} +import org.apache.spark.sql.connector.read.streaming.{ContinuousStream, MicroBatchStream} +import org.apache.spark.sql.internal.connector.SimpleTableProvider +import org.apache.spark.sql.sources.DataSourceRegister +import org.apache.spark.sql.types.{LongType, StructField, StructType, TimestampType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +/** + * A source that generates increment long values with timestamps. Each generated row has two + * columns: a timestamp column for the generated time and an auto increment long column starting + * with 0L. + * + * This source supports the following options: + * - `rowsPerMicroBatch` (e.g. 100): How many rows should be generated per micro-batch. + * - `numPartitions` (e.g. 10, default: Spark's default parallelism): The partition number for the + * generated rows. + * - `startTimestamp` (e.g. 1000, default: 0): starting value of generated time + * - `advanceMillisPerMicroBatch` (e.g. 1000, default: 1000): the amount of time being advanced in + * generated time on each micro-batch. + * + * Unlike `rate` data source, this data source provides a consistent set of input rows per + * micro-batch regardless of query execution (configuration of trigger, query being lagging, etc.), + * say, batch 0 will produce 0~999 and batch 1 will produce 1000~1999, and so on. Same applies to + * the generated time. + * + * As the name represents, this data source only supports micro-batch read. + */ +class RatePerMicroBatchProvider extends SimpleTableProvider with DataSourceRegister { + import RatePerMicroBatchProvider._ + + override def getTable(options: CaseInsensitiveStringMap): Table = { + val rowsPerBatch = options.getLong(ROWS_PER_BATCH, 0) + if (rowsPerBatch <= 0) { + throw new IllegalArgumentException( + s"Invalid value '$rowsPerBatch'. The option 'rowsPerBatch' must be positive") + } + + val numPartitions = options.getInt( + NUM_PARTITIONS, SparkSession.active.sparkContext.defaultParallelism) + if (numPartitions <= 0) { + throw new IllegalArgumentException( + s"Invalid value '$numPartitions'. The option 'numPartitions' must be positive") + } + + val startTimestamp = options.getLong(START_TIMESTAMP, 0) + if (startTimestamp < 0) { + throw new IllegalArgumentException( + s"Invalid value '$startTimestamp'. The option 'startTimestamp' must be non-negative") + } + + val advanceMillisPerBatch = options.getInt(ADVANCE_MILLIS_PER_BATCH, 1000) + if (advanceMillisPerBatch < 0) { Review comment: Technically it doesn't have any limitation. It can even be negative, though I can't think of the cases leveraging this. The reason I validate the value here is that I just wanted to block the case where the timestamp goes negative and keeps decreasing (the lower limit is startTimestamp = 0, advanceMillisPerBatch = 0). Having advanceMillisPerBatch = 0 doesn't lead to such issue, it just sticks the event time for all inputs across micro-batches. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
