[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r103894945 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala --- @@ -344,4 +346,36 @@ private[csv] object UnivocityParser { CSVUtils.filterCommentAndEmpty(linesWithoutHeader, options) filteredLines.flatMap(line => parser.parse(line)) } + + /** + * Parses a `Dataset` that contains CSV strings and turns it into an `RDD` of rows. + */ + def tokenizeDataset( + csvDataset: Dataset[String], + maybeFirstLine: Option[String], + options: CSVOptions): RDD[Array[String]] = { +val filtered = CSVUtils.filterCommentAndEmpty(csvDataset, options) +val linesWithoutHeader = maybeFirstLine.map { firstLine => + filtered.rdd.mapPartitions(CSVUtils.filterHeaderLine(_, firstLine, options)) +}.getOrElse(filtered.rdd) + +linesWithoutHeader.mapPartitions { iter => + val parser = new CsvParser(options.asParserSettings) + iter.map(line => parser.parseLine(line)) +} + } + + /** + * Parses a `Dataset` that contains CSV strings and turns it into an `RDD` of rows. + */ + def parseDataset( + csvDataset: Dataset[String], + schema: StructType, + maybeFirstLine: Option[String], + options: CSVOptions): RDD[InternalRow] = { +tokenizeDataset(csvDataset, maybeFirstLine, options).mapPartitions { iter => + val parser = new UnivocityParser(schema, options) + iter.flatMap(line => parser.convert(line)) +} + } --- End diff -- cc @cloud-fan, this is still a wip but I am trying to put the different execution paths into here in CSV parsing. For example, - `spark.read.csv(file)` - data: `parseIterator` (note that this one is read from partitioned file). - schema: `tokenizeDataset ` - `spark.read.csv(file)` with `wholeFile` - data: `parseStream` - schema: `tokenizeStream ` - `spark.read.csv(dataset)` - data: `parseDataset ` - schema: `tokenizeDataset ` However, it seems ending up with a bit weird arguments here.. do you think it is okay? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r103895193 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala --- @@ -344,4 +346,36 @@ private[csv] object UnivocityParser { CSVUtils.filterCommentAndEmpty(linesWithoutHeader, options) filteredLines.flatMap(line => parser.parse(line)) } + + /** + * Parses a `Dataset` that contains CSV strings and turns it into an `RDD` of rows. + */ + def tokenizeDataset( + csvDataset: Dataset[String], + maybeFirstLine: Option[String], + options: CSVOptions): RDD[Array[String]] = { +val filtered = CSVUtils.filterCommentAndEmpty(csvDataset, options) +val linesWithoutHeader = maybeFirstLine.map { firstLine => + filtered.rdd.mapPartitions(CSVUtils.filterHeaderLine(_, firstLine, options)) +}.getOrElse(filtered.rdd) + +linesWithoutHeader.mapPartitions { iter => + val parser = new CsvParser(options.asParserSettings) + iter.map(line => parser.parseLine(line)) +} + } + + /** + * Parses a `Dataset` that contains CSV strings and turns it into an `RDD` of rows. + */ + def parseDataset( + csvDataset: Dataset[String], + schema: StructType, + maybeFirstLine: Option[String], + options: CSVOptions): RDD[InternalRow] = { +tokenizeDataset(csvDataset, maybeFirstLine, options).mapPartitions { iter => + val parser = new UnivocityParser(schema, options) + iter.flatMap(line => parser.convert(line)) +} + } --- End diff -- If you are not sure or it looks painful to review, let me take out all the changes and put those into `DataFrameReader.csv` for now. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r104289988 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala --- @@ -399,6 +395,52 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging { } /** + * Loads an `Dataset[String]` storing CSV rows and returns the result as a `DataFrame`. + * + * Unless the schema is specified using `schema` function, this function goes through the + * input once to determine the input schema. + * + * @param csvDataset input Dataset with one CSV row per record + * @since 2.2.0 + */ + def csv(csvDataset: Dataset[String]): DataFrame = { +val parsedOptions: CSVOptions = new CSVOptions( + extraOptions.toMap, + sparkSession.sessionState.conf.sessionLocalTimeZone) +val filteredLines = CSVUtils.filterCommentAndEmpty(csvDataset, parsedOptions) +val maybeFirstLine = filteredLines.take(1).headOption +if (maybeFirstLine.isEmpty) { + return sparkSession.emptyDataFrame --- End diff -- The reason why this one exists unlike json is, CSV needs to head a head always first (even if it does not infer the schema, it needs at least the number of values). In this case, we could return empty one fast. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r104290005 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala --- @@ -399,6 +395,52 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging { } /** + * Loads an `Dataset[String]` storing CSV rows and returns the result as a `DataFrame`. + * + * Unless the schema is specified using `schema` function, this function goes through the + * input once to determine the input schema. + * + * @param csvDataset input Dataset with one CSV row per record + * @since 2.2.0 + */ + def csv(csvDataset: Dataset[String]): DataFrame = { +val parsedOptions: CSVOptions = new CSVOptions( + extraOptions.toMap, + sparkSession.sessionState.conf.sessionLocalTimeZone) +val filteredLines = CSVUtils.filterCommentAndEmpty(csvDataset, parsedOptions) +val maybeFirstLine = filteredLines.take(1).headOption +if (maybeFirstLine.isEmpty) { + return sparkSession.emptyDataFrame +} + +val firstLine = maybeFirstLine.get +val linesWithoutHeader: RDD[String] = filteredLines.rdd.mapPartitions( + CSVUtils.filterHeaderLine(_, firstLine, parsedOptions)) + +val schema = userSpecifiedSchema.getOrElse { --- End diff -- There is a similar code path in https://github.com/apache/spark/blob/7e5359be5ca038fdb579712b18e7f226d705c276/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala#L132-L150 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r104289951 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala --- @@ -604,6 +646,22 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging { } } + /** + * A convenient function for schema validation that takes `columnNameOfCorruptRecord` + * as an option. + */ + private def verifyColumnNameOfCorruptRecord( --- End diff -- Maybe, this is too much. I am willing to revert this back. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r104290082 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala --- @@ -604,6 +646,22 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging { } } + /** + * A convenient function for schema validation in datasources supporting + * `columnNameOfCorruptRecord` as an option. + */ + private def verifyColumnNameOfCorruptRecord( --- End diff -- Maybe, this is too much. I am willing to revert this back. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r104309644 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala --- @@ -59,11 +58,21 @@ abstract class CSVDataSource extends Serializable { sparkSession: SparkSession, inputPaths: Seq[FileStatus], parsedOptions: CSVOptions): Option[StructType] +} + +object CSVDataSource { + def apply(options: CSVOptions): CSVDataSource = { +if (options.wholeFile) { + WholeFileCSVDataSource +} else { + TextInputCSVDataSource +} + } /** * Generates a header from the given row which is null-safe and duplicate-safe. */ - protected def makeSafeHeader( + def makeSafeHeader( --- End diff -- `makeSafeHeader` was moved from `CSVDataSource` class to `CSVDataSource` companion object so that this can be accessed in `DataFrameReader`. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16854: [WIP][SPARK-15463][SQL] Add an API to load DataFr...
Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/16854#discussion_r100229312 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala --- @@ -361,6 +362,41 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging { } /** + * Loads an `Dataset[String]` storing CSV rows and returns the result as a `DataFrame`. + * + * Unless the schema is specified using `schema` function, this function goes through the + * input once to determine the input schema. + * + * @param csvDataset input Dataset with one CSV row per record + * @since 2.2.0 + */ + def csv(csvDataset: Dataset[String]): DataFrame = { +val parsedOptions: CSVOptions = new CSVOptions(extraOptions.toMap) --- End diff -- Just to help review, there is a similar code path in https://github.com/apache/spark/blob/3d314d08c9420e74b4bb687603cdd11394eccab5/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala#L105-L125 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org