[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_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_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_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_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_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_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
