You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by MaxGekk <gi...@git.apache.org> on 2018/05/10 18:56:40 UTC
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
GitHub user MaxGekk opened a pull request:
https://github.com/apache/spark/pull/21296
[SPARK-24244][SQL] CSV column pruning
## What changes were proposed in this pull request?
uniVocity parser allows to specify only required column names or indexes for [parsing](https://www.univocity.com/pages/parsers-tutorial) like:
```
// Here we select only the columns by their indexes.
// The parser just skips the values in other columns
parserSettings.selectIndexes(4, 0, 1);
CsvParser parser = new CsvParser(parserSettings);
```
In this PR, I propose to extract indexes from required schema and pass them into the CSV parser. Benchmarks show the following improvements in parsing of 1000 columns:
```
Select 100 columns out of 1000: x1.76
Select 1 column out of 1000: x2
```
**Note**: Comparing to current implementation, the changes can return different result for malformed rows if only subset of all columns is requested.
## How was this patch tested?
It was tested by new test which selects 3 columns out of 15, by existing tests and by new benchmarks.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/MaxGekk/spark-1 csv-column-pruning
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/21296.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #21296
----
commit 9cffa0fccc33552e8fce3580a9a665b022f5bf22
Author: Maxim Gekk <ma...@...>
Date: 2018-03-21T20:03:11Z
Adding tests for select only requested columns
commit fdbcbe3536aee04e6a84b72ac319726614416bc3
Author: Maxim Gekk <ma...@...>
Date: 2018-03-21T20:42:08Z
Select indexes of required columns only
commit 578f47b0f32a76caf6c9ede8763c9cf85a1c83e9
Author: Maxim Gekk <ma...@...>
Date: 2018-03-24T10:41:29Z
Fix the case when number of parsed fields are not matched to required schema
commit 0f942c308dca173dad8f421e893066b8c03d35a3
Author: Maxim Gekk <ma...@...>
Date: 2018-03-24T11:07:55Z
Using selectIndexes if required number of columns are less than its total number.
commit c4b11601e9c264729e141fff3dc653d868a7ad69
Author: Maxim Gekk <ma...@...>
Date: 2018-03-24T11:48:43Z
Fix the test: force to read all columns
commit 8cf6eab952d79628cb8ee2ff7b92dadae60ec686
Author: Maxim Gekk <ma...@...>
Date: 2018-04-06T20:55:35Z
Fix merging conflicts
commit 5b2f0b9d7346f927842bc1a2089a7299876f1894
Author: Maxim Gekk <ma...@...>
Date: 2018-04-29T11:52:08Z
Benchmarks for many columns
commit 6d1e902c0011e88dbafb65c4ad6e7431370ed12d
Author: Maxim Gekk <ma...@...>
Date: 2018-04-29T12:59:58Z
Make size of requiredSchema equals to amount of selected columns
commit 4525795f7337cbd081f569cd79d7f90cb58edbee
Author: Maxim Gekk <ma...@...>
Date: 2018-04-29T13:36:54Z
Removing selection of all columns
commit 8809cecf93d8e7a97eca827d9e8637a7eb5b2449
Author: Maxim Gekk <ma...@...>
Date: 2018-04-29T13:50:44Z
Updating benchmarks for select indexes
commit dc97ceb96185ed2eaa05fbe1aee8ecfe8ccb7e7d
Author: Maxim Gekk <ma...@...>
Date: 2018-05-05T19:19:17Z
Addressing Herman's review comments
commit 51b31483263e13cd85b19b3efea65188945eda99
Author: Maxim Gekk <ma...@...>
Date: 2018-05-10T18:39:38Z
Updated benchmark result for recent changes
commit e3958b1468b490b548574b53512f0d83850e6f6f
Author: Maxim Gekk <ma...@...>
Date: 2018-05-10T18:46:17Z
Add ticket number to test title
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187832513
--- Diff: docs/sql-programming-guide.md ---
@@ -1814,6 +1814,7 @@ working with timestamps in `pandas_udf`s to get the best performance, see
- In version 2.3 and earlier, `to_utc_timestamp` and `from_utc_timestamp` respect the timezone in the input timestamp string, which breaks the assumption that the input timestamp is in a specific timezone. Therefore, these 2 functions can return unexpected results. In version 2.4 and later, this problem has been fixed. `to_utc_timestamp` and `from_utc_timestamp` will return null if the input timestamp string contains timezone. As an example, `from_utc_timestamp('2000-10-10 00:00:00', 'GMT+1')` will return `2000-10-10 01:00:00` in both Spark 2.3 and 2.4. However, `from_utc_timestamp('2000-10-10 00:00:00+00:00', 'GMT+1')`, assuming a local timezone of GMT+8, will return `2000-10-10 09:00:00` in Spark 2.3 but `null` in 2.4. For people who don't care about this problem and want to retain the previous behaivor to keep their query unchanged, you can set `spark.sql.function.rejectTimezoneInString` to false. This option will be removed in Spark 3.0 and should only be used as a tempora
ry workaround.
- In version 2.3 and earlier, Spark converts Parquet Hive tables by default but ignores table properties like `TBLPROPERTIES (parquet.compression 'NONE')`. This happens for ORC Hive table properties like `TBLPROPERTIES (orc.compress 'NONE')` in case of `spark.sql.hive.convertMetastoreOrc=true`, too. Since Spark 2.4, Spark respects Parquet/ORC specific table properties while converting Parquet/ORC Hive tables. As an example, `CREATE TABLE t(id int) STORED AS PARQUET TBLPROPERTIES (parquet.compression 'NONE')` would generate Snappy parquet files during insertion in Spark 2.3, and in Spark 2.4, the result would be uncompressed parquet files.
- Since Spark 2.0, Spark converts Parquet Hive tables by default for better performance. Since Spark 2.4, Spark converts ORC Hive tables by default, too. It means Spark uses its own ORC support by default instead of Hive SerDe. As an example, `CREATE TABLE t(id int) STORED AS ORC` would be handled with Hive SerDe in Spark 2.3, and in Spark 2.4, it would be converted into Spark's ORC data source table and ORC vectorization would be applied. To set `false` to `spark.sql.hive.convertMetastoreOrc` restores the previous behavior.
+ - Since Spark 2.4, handling of malformed rows in CSV files was changed. Previously, all column values of every row are parsed independently of its future usage. A row was considered as malformed if the CSV parser wasn't able to handle any column value in the row even if the value wasn't requested. Starting from version 2.4, only requested column values are parsed, and other values can be ignored. In such way, correct column values that were considered as malformed in previous Spark version only because of other malformed values become correct in Spark version 2.4.
--- End diff --
can we follow the style of other migration guides?
```
In version 2.3 and earlier, xxxx. Since Spark 2.4, xxxx. As an example, xxxx. (and talk about the flag to restore the previous behavior)
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21296
I'm fine with breaking the CSV behavior of malformed input a little, as long as we documented it well in the migration guide. cc @gatorsmile
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90471 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90471/testReport)** for PR 21296 at commit [`e3958b1`](https://github.com/apache/spark/commit/e3958b1468b490b548574b53512f0d83850e6f6f).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90471 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90471/testReport)** for PR 21296 at commit [`e3958b1`](https://github.com/apache/spark/commit/e3958b1468b490b548574b53512f0d83850e6f6f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90751/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90797 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90797/testReport)** for PR 21296 at commit [`6ff6d4f`](https://github.com/apache/spark/commit/6ff6d4fda9f7e8ee43d7aa04818204de4c49440b).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on the issue:
https://github.com/apache/spark/pull/21296
Can we update the migration guide then? I want to see if the note makes sense.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187500240
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala ---
@@ -267,7 +267,7 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
.options(Map("header" -> "true", "mode" -> "dropmalformed"))
.load(testFile(carsFile))
- assert(cars.select("year").collect().size === 2)
+ assert(cars.collect().size === 2)
--- End diff --
This changes behaviour and it's intendedly parsed to keep the backword compatibility. There was an issue about the different number of counts. I think you are basically saying `cars.select("year").collect().size` and `cars.collect().size` are different and they are correct, right?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90751 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90751/testReport)** for PR 21296 at commit [`7dcfc7a`](https://github.com/apache/spark/commit/7dcfc7a7664fcd5311cb352f0ea7a24b3cc1c639).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/21296
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on the issue:
https://github.com/apache/spark/pull/21296
@cloud-fan @hvanhovell Could you look at the PR, please.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187608568
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVBenchmarks.scala ---
@@ -0,0 +1,92 @@
+/*
+ * 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.datasources.csv
+
+import java.io.File
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Column, Row, SparkSession}
+import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.types._
+import org.apache.spark.util.{Benchmark, Utils}
+
+/**
+ * Benchmark to measure CSV read/write performance.
+ * To run this:
+ * spark-submit --class <this class> --jars <spark sql test jar>
+ */
+object CSVBenchmarks {
+ val conf = new SparkConf()
+
+ val spark = SparkSession.builder
+ .master("local[1]")
+ .appName("benchmark-csv-datasource")
+ .config(conf)
+ .getOrCreate()
+ import spark.implicits._
+
+ def withTempPath(f: File => Unit): Unit = {
+ val path = Utils.createTempDir()
+ path.delete()
+ try f(path) finally Utils.deleteRecursively(path)
+ }
+
+ def multiColumnsBenchmark(rowsNum: Int): Unit = {
+ val colsNum = 1000
+ val benchmark = new Benchmark(s"Wide rows with $colsNum columns", rowsNum)
+
+ withTempPath { path =>
+ val fields = Seq.tabulate(colsNum)(i => StructField(s"col$i", IntegerType))
+ val schema = StructType(fields)
+ val values = (0 until colsNum).map(i => i.toString).mkString(",")
+ val columnNames = schema.fieldNames
+
+ spark.range(rowsNum)
+ .select(Seq.tabulate(colsNum)(i => lit(i).as(s"col$i")): _*)
+ .write.option("header", true)
+ .csv(path.getAbsolutePath)
+
+ val ds = spark.read.schema(schema).csv(path.getAbsolutePath)
+
+ benchmark.addCase(s"Select $colsNum columns", 3) { _ =>
+ ds.select("*").filter((row: Row) => true).count()
+ }
+ val cols100 = columnNames.take(100).map(Column(_))
+ benchmark.addCase(s"Select 100 columns", 3) { _ =>
+ ds.select(cols100: _*).filter((row: Row) => true).count()
+ }
+ benchmark.addCase(s"Select one column", 3) { _ =>
+ ds.select($"col1").filter((row: Row) => true).count()
+ }
+
+ /*
+ Intel(R) Core(TM) i7-7920HQ CPU @ 3.10GHz
+
+ Wide rows with 1000 columns: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
+ --------------------------------------------------------------------------------------------
+ Select 1000 columns 76910 / 78065 0.0 76909.8 1.0X
+ Select 100 columns 28625 / 32884 0.0 28625.1 2.7X
+ Select one column 22498 / 22669 0.0 22497.8 3.4X
--- End diff --
You are right, avoiding unnecessary casting speeds up more than 2 times. We can see that on this benchmark before my changes. Without the changes, selecting only one string column takes [44.5 seconds](https://github.com/apache/spark/pull/21296/commits/8809cecf93d8e7a97eca827d9e8637a7eb5b2449#diff-985fa5181f2aec4df39324995590ea83L114) but select of all columns ~80 seconds.
> ... relatively expensive comparing to the parsing logic.
As the benchmark shows we can achieve performance improvements in parsing too. Selecting only 1 out of 1000 columns takes 22.5 seconds but without the PR it takes 44.5:
https://github.com/apache/spark/pull/21296/commits/8809cecf93d8e7a97eca827d9e8637a7eb5b2449
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90751 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90751/testReport)** for PR 21296 at commit [`7dcfc7a`](https://github.com/apache/spark/commit/7dcfc7a7664fcd5311cb352f0ea7a24b3cc1c639).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* ` case class WidenSetOperationTypes(conf: SQLConf) extends Rule[LogicalPlan] `
* ` case class FunctionArgumentConversion(conf: SQLConf) extends TypeCoercionRule `
* ` case class CaseWhenCoercion(conf: SQLConf) extends TypeCoercionRule `
* ` case class IfCoercion(conf: SQLConf) extends TypeCoercionRule `
* ` case class ImplicitTypeCasts(conf: SQLConf) extends TypeCoercionRule `
* `class ContinuousMemoryStream[A : Encoder](id: Int, sqlContext: SQLContext, numPartitions: Int = 2)`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by koertkuipers <gi...@git.apache.org>.
Github user koertkuipers commented on the issue:
https://github.com/apache/spark/pull/21296
if i do not select a schema (and i use inferSchema), and i do a select for only a few column, does this push down the column selection into the reading of data (for schema inference and for the actual data read)?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90555 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90555/testReport)** for PR 21296 at commit [`f90daa7`](https://github.com/apache/spark/commit/f90daa7ea33d119be978c27de10978c2d6281e25).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90555/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187604963
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala ---
@@ -267,7 +267,7 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
.options(Map("header" -> "true", "mode" -> "dropmalformed"))
.load(testFile(carsFile))
- assert(cars.select("year").collect().size === 2)
+ assert(cars.collect().size === 2)
--- End diff --
> it's intendedly parsed to keep the backword compatibility.
Right, by selecting all columns I force *UnivocityParser* to fall to the case:
https://github.com/MaxGekk/spark-1/blob/a4a0a549156a15011c33c7877a35f244d75b7a4f/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala#L193-L213
when number of returned tokens are less than required.
In the case of `cars.select("year")`, uniVocity parser returns only one token as it is expected.
> There was an issue about the different number of counts.
The PR changes behavior for some malformed inputs but I believe we could provide better performance for users who have correct inputs.
> I think you are basically saying cars.select("year").collect().size and cars.collect().size are different and they are correct, right?
Yes, you can say that. You are right it seems the PR proposes another interpretation for malformed rows. `cars.select("year")` is:
```
+----+
|year|
+----+
|2012|
|1997|
|2015|
+----+
```
and we should not reject `2015` only because there are problems in not requested columns. In this particular case, the last row consists of only one value at `0` position and it is correct.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r189277890
--- Diff: docs/sql-programming-guide.md ---
@@ -1814,6 +1814,7 @@ working with timestamps in `pandas_udf`s to get the best performance, see
- In version 2.3 and earlier, `to_utc_timestamp` and `from_utc_timestamp` respect the timezone in the input timestamp string, which breaks the assumption that the input timestamp is in a specific timezone. Therefore, these 2 functions can return unexpected results. In version 2.4 and later, this problem has been fixed. `to_utc_timestamp` and `from_utc_timestamp` will return null if the input timestamp string contains timezone. As an example, `from_utc_timestamp('2000-10-10 00:00:00', 'GMT+1')` will return `2000-10-10 01:00:00` in both Spark 2.3 and 2.4. However, `from_utc_timestamp('2000-10-10 00:00:00+00:00', 'GMT+1')`, assuming a local timezone of GMT+8, will return `2000-10-10 09:00:00` in Spark 2.3 but `null` in 2.4. For people who don't care about this problem and want to retain the previous behaivor to keep their query unchanged, you can set `spark.sql.function.rejectTimezoneInString` to false. This option will be removed in Spark 3.0 and should only be used as a tempora
ry workaround.
- In version 2.3 and earlier, Spark converts Parquet Hive tables by default but ignores table properties like `TBLPROPERTIES (parquet.compression 'NONE')`. This happens for ORC Hive table properties like `TBLPROPERTIES (orc.compress 'NONE')` in case of `spark.sql.hive.convertMetastoreOrc=true`, too. Since Spark 2.4, Spark respects Parquet/ORC specific table properties while converting Parquet/ORC Hive tables. As an example, `CREATE TABLE t(id int) STORED AS PARQUET TBLPROPERTIES (parquet.compression 'NONE')` would generate Snappy parquet files during insertion in Spark 2.3, and in Spark 2.4, the result would be uncompressed parquet files.
- Since Spark 2.0, Spark converts Parquet Hive tables by default for better performance. Since Spark 2.4, Spark converts ORC Hive tables by default, too. It means Spark uses its own ORC support by default instead of Hive SerDe. As an example, `CREATE TABLE t(id int) STORED AS ORC` would be handled with Hive SerDe in Spark 2.3, and in Spark 2.4, it would be converted into Spark's ORC data source table and ORC vectorization would be applied. To set `false` to `spark.sql.hive.convertMetastoreOrc` restores the previous behavior.
+ - In version 2.3 and earlier, CSV rows are considered as malformed if at least one column value in the row is malformed. CSV parser dropped such rows in the DROPMALFORMED mode or outputs an error in the FAILFAST mode. Since Spark 2.4, CSV row is considered as malformed only when it contains malformed column values requested from CSV datasource, other values can be ignored. As an example, CSV file contains the "id,name" header and one row "1234". In Spark 2.4, selection of the id column consists of a row with one column value 1234 but in Spark 2.3 and earlier it is empty in the DROPMALFORMED mode. To restore the previous behavior, all column values must be requested. This example demonstrates how to achieve that with filter in Scala `spark.read.option("header", true).option("mode", "dropmalformed").csv("a.csv").filter(_ => true).select("id")`.
--- End diff --
I am going to add `spark.sql.csv.parser.columnPruning.enabled` to SQLConf. Is it ok for you?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90473 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90473/testReport)** for PR 21296 at commit [`a4a0a54`](https://github.com/apache/spark/commit/a4a0a549156a15011c33c7877a35f244d75b7a4f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187610192
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala ---
@@ -73,11 +64,24 @@ class UnivocityParser(
// Each input token is placed in each output row's position by mapping these. In this case,
//
// output row - ["A", 2]
- private val valueConverters: Array[ValueConverter] =
- schema.map(f => makeConverter(f.name, f.dataType, f.nullable, options)).toArray
+ private val valueConverters: Array[ValueConverter] = {
+ requiredSchema.map(f => makeConverter(f.name, f.dataType, f.nullable, options)).toArray
+ }
- private val tokenIndexArr: Array[Int] = {
- requiredSchema.map(f => schema.indexOf(f)).toArray
+ private val tokenizer = {
+ val parserSetting = options.asParserSettings
+ if (requiredSchema.length < schema.length) {
+ val tokenIndexArr = requiredSchema.map(f => java.lang.Integer.valueOf(schema.indexOf(f)))
+ parserSetting.selectIndexes(tokenIndexArr: _*)
--- End diff --
I tried the changes as an experiment because some of our clients have many columns (> 200 columns) in their input CSV files. The experiment shows that the improvements can significantly impact on total execution time.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187499797
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVBenchmarks.scala ---
@@ -0,0 +1,92 @@
+/*
+ * 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.datasources.csv
+
+import java.io.File
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Column, Row, SparkSession}
+import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.types._
+import org.apache.spark.util.{Benchmark, Utils}
+
+/**
+ * Benchmark to measure CSV read/write performance.
+ * To run this:
+ * spark-submit --class <this class> --jars <spark sql test jar>
+ */
+object CSVBenchmarks {
+ val conf = new SparkConf()
+
+ val spark = SparkSession.builder
+ .master("local[1]")
+ .appName("benchmark-csv-datasource")
+ .config(conf)
+ .getOrCreate()
+ import spark.implicits._
+
+ def withTempPath(f: File => Unit): Unit = {
+ val path = Utils.createTempDir()
+ path.delete()
+ try f(path) finally Utils.deleteRecursively(path)
+ }
+
+ def multiColumnsBenchmark(rowsNum: Int): Unit = {
+ val colsNum = 1000
+ val benchmark = new Benchmark(s"Wide rows with $colsNum columns", rowsNum)
+
+ withTempPath { path =>
+ val fields = Seq.tabulate(colsNum)(i => StructField(s"col$i", IntegerType))
+ val schema = StructType(fields)
+ val values = (0 until colsNum).map(i => i.toString).mkString(",")
+ val columnNames = schema.fieldNames
+
+ spark.range(rowsNum)
+ .select(Seq.tabulate(colsNum)(i => lit(i).as(s"col$i")): _*)
+ .write.option("header", true)
+ .csv(path.getAbsolutePath)
+
+ val ds = spark.read.schema(schema).csv(path.getAbsolutePath)
+
+ benchmark.addCase(s"Select $colsNum columns", 3) { _ =>
+ ds.select("*").filter((row: Row) => true).count()
+ }
+ val cols100 = columnNames.take(100).map(Column(_))
+ benchmark.addCase(s"Select 100 columns", 3) { _ =>
+ ds.select(cols100: _*).filter((row: Row) => true).count()
+ }
+ benchmark.addCase(s"Select one column", 3) { _ =>
+ ds.select($"col1").filter((row: Row) => true).count()
+ }
+
+ /*
+ Intel(R) Core(TM) i7-7920HQ CPU @ 3.10GHz
+
+ Wide rows with 1000 columns: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
+ --------------------------------------------------------------------------------------------
+ Select 1000 columns 76910 / 78065 0.0 76909.8 1.0X
+ Select 100 columns 28625 / 32884 0.0 28625.1 2.7X
+ Select one column 22498 / 22669 0.0 22497.8 3.4X
--- End diff --
BTW, I think we are already doing the column pruning by avoiding casting cost which is relatively expensive comparing to the parsing logic.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90797 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90797/testReport)** for PR 21296 at commit [`6ff6d4f`](https://github.com/apache/spark/commit/6ff6d4fda9f7e8ee43d7aa04818204de4c49440b).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187499921
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala ---
@@ -73,11 +64,24 @@ class UnivocityParser(
// Each input token is placed in each output row's position by mapping these. In this case,
//
// output row - ["A", 2]
- private val valueConverters: Array[ValueConverter] =
- schema.map(f => makeConverter(f.name, f.dataType, f.nullable, options)).toArray
+ private val valueConverters: Array[ValueConverter] = {
+ requiredSchema.map(f => makeConverter(f.name, f.dataType, f.nullable, options)).toArray
+ }
- private val tokenIndexArr: Array[Int] = {
- requiredSchema.map(f => schema.indexOf(f)).toArray
+ private val tokenizer = {
+ val parserSetting = options.asParserSettings
+ if (requiredSchema.length < schema.length) {
+ val tokenIndexArr = requiredSchema.map(f => java.lang.Integer.valueOf(schema.indexOf(f)))
+ parserSetting.selectIndexes(tokenIndexArr: _*)
--- End diff --
I think I tried this locally but I didn't submit a PR since the improvement was trivial and a test was broken fwiw.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on the issue:
https://github.com/apache/spark/pull/21296
mind fixing the PR title? sounds we have never implemented the "column pruning" before in CSV.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on the issue:
https://github.com/apache/spark/pull/21296
@cloud-fan @HyukjinKwon Could you look at the PR, please.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on the issue:
https://github.com/apache/spark/pull/21296
> ... does this push down the column selection into the reading of data
Yes, it does.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] CSV column pruning
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187426203
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala ---
@@ -267,7 +267,7 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
.options(Map("header" -> "true", "mode" -> "dropmalformed"))
.load(testFile(carsFile))
- assert(cars.select("year").collect().size === 2)
+ assert(cars.collect().size === 2)
--- End diff --
The `cars.csv` file has header with 5 columns:
```
year,make,model,comment,blank
```
and 2 rows with 4 valid columns and the last one is blank:
```
"2012","Tesla","S","No comment",
1997,Ford,E350,"Go get one now they are going fast",
```
and one more row with only with 3 columns:
```
2015,Chevy,Volt
```
Previous (current) implementation drops the last row in the `dropmalformed` mode because it parses whole rows, and the last one is incorrect. If only the `year` column is selected, uniVocity parser returns values for first column (with index `0`) and doesn't analyze correctness of the rest part of the rows. So in this way `cars.select("year").collect().size` returns `3`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r189898241
--- Diff: docs/sql-programming-guide.md ---
@@ -1814,6 +1814,7 @@ working with timestamps in `pandas_udf`s to get the best performance, see
- In version 2.3 and earlier, `to_utc_timestamp` and `from_utc_timestamp` respect the timezone in the input timestamp string, which breaks the assumption that the input timestamp is in a specific timezone. Therefore, these 2 functions can return unexpected results. In version 2.4 and later, this problem has been fixed. `to_utc_timestamp` and `from_utc_timestamp` will return null if the input timestamp string contains timezone. As an example, `from_utc_timestamp('2000-10-10 00:00:00', 'GMT+1')` will return `2000-10-10 01:00:00` in both Spark 2.3 and 2.4. However, `from_utc_timestamp('2000-10-10 00:00:00+00:00', 'GMT+1')`, assuming a local timezone of GMT+8, will return `2000-10-10 09:00:00` in Spark 2.3 but `null` in 2.4. For people who don't care about this problem and want to retain the previous behaivor to keep their query unchanged, you can set `spark.sql.function.rejectTimezoneInString` to false. This option will be removed in Spark 3.0 and should only be used as a tempora
ry workaround.
- In version 2.3 and earlier, Spark converts Parquet Hive tables by default but ignores table properties like `TBLPROPERTIES (parquet.compression 'NONE')`. This happens for ORC Hive table properties like `TBLPROPERTIES (orc.compress 'NONE')` in case of `spark.sql.hive.convertMetastoreOrc=true`, too. Since Spark 2.4, Spark respects Parquet/ORC specific table properties while converting Parquet/ORC Hive tables. As an example, `CREATE TABLE t(id int) STORED AS PARQUET TBLPROPERTIES (parquet.compression 'NONE')` would generate Snappy parquet files during insertion in Spark 2.3, and in Spark 2.4, the result would be uncompressed parquet files.
- Since Spark 2.0, Spark converts Parquet Hive tables by default for better performance. Since Spark 2.4, Spark converts ORC Hive tables by default, too. It means Spark uses its own ORC support by default instead of Hive SerDe. As an example, `CREATE TABLE t(id int) STORED AS ORC` would be handled with Hive SerDe in Spark 2.3, and in Spark 2.4, it would be converted into Spark's ORC data source table and ORC vectorization would be applied. To set `false` to `spark.sql.hive.convertMetastoreOrc` restores the previous behavior.
+ - In version 2.3 and earlier, CSV rows are considered as malformed if at least one column value in the row is malformed. CSV parser dropped such rows in the DROPMALFORMED mode or outputs an error in the FAILFAST mode. Since Spark 2.4, CSV row is considered as malformed only when it contains malformed column values requested from CSV datasource, other values can be ignored. As an example, CSV file contains the "id,name" header and one row "1234". In Spark 2.4, selection of the id column consists of a row with one column value 1234 but in Spark 2.3 and earlier it is empty in the DROPMALFORMED mode. To restore the previous behavior, all column values must be requested. This example demonstrates how to achieve that with filter in Scala `spark.read.option("header", true).option("mode", "dropmalformed").csv("a.csv").filter(_ => true).select("id")`.
--- End diff --
shall we follow the parquet configs and just use `spark.sql.csv.columnPruning.enabled`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90797/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187823861
--- Diff: docs/sql-programming-guide.md ---
@@ -1814,6 +1814,7 @@ working with timestamps in `pandas_udf`s to get the best performance, see
- In version 2.3 and earlier, `to_utc_timestamp` and `from_utc_timestamp` respect the timezone in the input timestamp string, which breaks the assumption that the input timestamp is in a specific timezone. Therefore, these 2 functions can return unexpected results. In version 2.4 and later, this problem has been fixed. `to_utc_timestamp` and `from_utc_timestamp` will return null if the input timestamp string contains timezone. As an example, `from_utc_timestamp('2000-10-10 00:00:00', 'GMT+1')` will return `2000-10-10 01:00:00` in both Spark 2.3 and 2.4. However, `from_utc_timestamp('2000-10-10 00:00:00+00:00', 'GMT+1')`, assuming a local timezone of GMT+8, will return `2000-10-10 09:00:00` in Spark 2.3 but `null` in 2.4. For people who don't care about this problem and want to retain the previous behaivor to keep their query unchanged, you can set `spark.sql.function.rejectTimezoneInString` to false. This option will be removed in Spark 3.0 and should only be used as a tempora
ry workaround.
- In version 2.3 and earlier, Spark converts Parquet Hive tables by default but ignores table properties like `TBLPROPERTIES (parquet.compression 'NONE')`. This happens for ORC Hive table properties like `TBLPROPERTIES (orc.compress 'NONE')` in case of `spark.sql.hive.convertMetastoreOrc=true`, too. Since Spark 2.4, Spark respects Parquet/ORC specific table properties while converting Parquet/ORC Hive tables. As an example, `CREATE TABLE t(id int) STORED AS PARQUET TBLPROPERTIES (parquet.compression 'NONE')` would generate Snappy parquet files during insertion in Spark 2.3, and in Spark 2.4, the result would be uncompressed parquet files.
- Since Spark 2.0, Spark converts Parquet Hive tables by default for better performance. Since Spark 2.4, Spark converts ORC Hive tables by default, too. It means Spark uses its own ORC support by default instead of Hive SerDe. As an example, `CREATE TABLE t(id int) STORED AS ORC` would be handled with Hive SerDe in Spark 2.3, and in Spark 2.4, it would be converted into Spark's ORC data source table and ORC vectorization would be applied. To set `false` to `spark.sql.hive.convertMetastoreOrc` restores the previous behavior.
+ - Since Spark 2.4, handling of malformed rows in CSV files was changed. Previously, all column values of every row are parsed independently of its future usage. A row was considered as malformed if the CSV parser wasn't able to handle any column value in the row even if the value wasn't requested. Starting from version 2.4, only requested column values are parsed, and other values can be ignored. In such way, correct column values that were considered as malformed in previous Spark version only because of other malformed values become correct in Spark version 2.4.
--- End diff --
Do we really wanna this behaviour change @cloud-fan and @gatorsmile?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on the issue:
https://github.com/apache/spark/pull/21296
I added the word `parser` to the feature name because as @HyukjinKwon wrote above we do pruning in type conversion already. This PR enables column pruning by CSV parser only.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90752 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90752/testReport)** for PR 21296 at commit [`f89eeb7`](https://github.com/apache/spark/commit/f89eeb7f7ba86888ad3f7994577a4d4ebbf09197).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r189195575
--- Diff: docs/sql-programming-guide.md ---
@@ -1814,6 +1814,7 @@ working with timestamps in `pandas_udf`s to get the best performance, see
- In version 2.3 and earlier, `to_utc_timestamp` and `from_utc_timestamp` respect the timezone in the input timestamp string, which breaks the assumption that the input timestamp is in a specific timezone. Therefore, these 2 functions can return unexpected results. In version 2.4 and later, this problem has been fixed. `to_utc_timestamp` and `from_utc_timestamp` will return null if the input timestamp string contains timezone. As an example, `from_utc_timestamp('2000-10-10 00:00:00', 'GMT+1')` will return `2000-10-10 01:00:00` in both Spark 2.3 and 2.4. However, `from_utc_timestamp('2000-10-10 00:00:00+00:00', 'GMT+1')`, assuming a local timezone of GMT+8, will return `2000-10-10 09:00:00` in Spark 2.3 but `null` in 2.4. For people who don't care about this problem and want to retain the previous behaivor to keep their query unchanged, you can set `spark.sql.function.rejectTimezoneInString` to false. This option will be removed in Spark 3.0 and should only be used as a tempora
ry workaround.
- In version 2.3 and earlier, Spark converts Parquet Hive tables by default but ignores table properties like `TBLPROPERTIES (parquet.compression 'NONE')`. This happens for ORC Hive table properties like `TBLPROPERTIES (orc.compress 'NONE')` in case of `spark.sql.hive.convertMetastoreOrc=true`, too. Since Spark 2.4, Spark respects Parquet/ORC specific table properties while converting Parquet/ORC Hive tables. As an example, `CREATE TABLE t(id int) STORED AS PARQUET TBLPROPERTIES (parquet.compression 'NONE')` would generate Snappy parquet files during insertion in Spark 2.3, and in Spark 2.4, the result would be uncompressed parquet files.
- Since Spark 2.0, Spark converts Parquet Hive tables by default for better performance. Since Spark 2.4, Spark converts ORC Hive tables by default, too. It means Spark uses its own ORC support by default instead of Hive SerDe. As an example, `CREATE TABLE t(id int) STORED AS ORC` would be handled with Hive SerDe in Spark 2.3, and in Spark 2.4, it would be converted into Spark's ORC data source table and ORC vectorization would be applied. To set `false` to `spark.sql.hive.convertMetastoreOrc` restores the previous behavior.
+ - In version 2.3 and earlier, CSV rows are considered as malformed if at least one column value in the row is malformed. CSV parser dropped such rows in the DROPMALFORMED mode or outputs an error in the FAILFAST mode. Since Spark 2.4, CSV row is considered as malformed only when it contains malformed column values requested from CSV datasource, other values can be ignored. As an example, CSV file contains the "id,name" header and one row "1234". In Spark 2.4, selection of the id column consists of a row with one column value 1234 but in Spark 2.3 and earlier it is empty in the DROPMALFORMED mode. To restore the previous behavior, all column values must be requested. This example demonstrates how to achieve that with filter in Scala `spark.read.option("header", true).option("mode", "dropmalformed").csv("a.csv").filter(_ => true).select("id")`.
--- End diff --
instead of using the `filter(_ => true)` trick, can we provide a config?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90471/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/21296
This PR breaks the tests. Let me revert it. Please resubmit the PR with the fix. Thanks!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/21296
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/91045/testReport/org.apache.spark.sql.execution.datasources.csv/UnivocityParserSuite/_It_is_not_a_test_it_is_a_sbt_testing_SuiteSelector_/history/
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187810542
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala ---
@@ -1322,4 +1322,31 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
val sampled = spark.read.option("inferSchema", true).option("samplingRatio", 1.0).csv(ds)
assert(sampled.count() == ds.count())
}
+
+ test("SPARK-24244: Select a little of many columns") {
--- End diff --
I added the test to check that requesting only subset of all columns works correctly. And to check the case when ordering of fields in required schema is different from the data schema. Previously I had a concern that if I select columns in different order like `select('f15, 'f10, 'f5)`, I will get the required schema with the same field order. It seems the required schema has the same order as data schema. That's why I removed https://github.com/apache/spark/pull/21296/files/a4a0a549156a15011c33c7877a35f244d75b7a4f#diff-d19881aceddcaa5c60620fdcda99b4c4L214
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187675312
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala ---
@@ -1322,4 +1322,31 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
val sampled = spark.read.option("inferSchema", true).option("samplingRatio", 1.0).csv(ds)
assert(sampled.count() == ds.count())
}
+
+ test("SPARK-24244: Select a little of many columns") {
--- End diff --
what does this test?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187823801
--- Diff: docs/sql-programming-guide.md ---
@@ -1814,6 +1814,7 @@ working with timestamps in `pandas_udf`s to get the best performance, see
- In version 2.3 and earlier, `to_utc_timestamp` and `from_utc_timestamp` respect the timezone in the input timestamp string, which breaks the assumption that the input timestamp is in a specific timezone. Therefore, these 2 functions can return unexpected results. In version 2.4 and later, this problem has been fixed. `to_utc_timestamp` and `from_utc_timestamp` will return null if the input timestamp string contains timezone. As an example, `from_utc_timestamp('2000-10-10 00:00:00', 'GMT+1')` will return `2000-10-10 01:00:00` in both Spark 2.3 and 2.4. However, `from_utc_timestamp('2000-10-10 00:00:00+00:00', 'GMT+1')`, assuming a local timezone of GMT+8, will return `2000-10-10 09:00:00` in Spark 2.3 but `null` in 2.4. For people who don't care about this problem and want to retain the previous behaivor to keep their query unchanged, you can set `spark.sql.function.rejectTimezoneInString` to false. This option will be removed in Spark 3.0 and should only be used as a tempora
ry workaround.
- In version 2.3 and earlier, Spark converts Parquet Hive tables by default but ignores table properties like `TBLPROPERTIES (parquet.compression 'NONE')`. This happens for ORC Hive table properties like `TBLPROPERTIES (orc.compress 'NONE')` in case of `spark.sql.hive.convertMetastoreOrc=true`, too. Since Spark 2.4, Spark respects Parquet/ORC specific table properties while converting Parquet/ORC Hive tables. As an example, `CREATE TABLE t(id int) STORED AS PARQUET TBLPROPERTIES (parquet.compression 'NONE')` would generate Snappy parquet files during insertion in Spark 2.3, and in Spark 2.4, the result would be uncompressed parquet files.
- Since Spark 2.0, Spark converts Parquet Hive tables by default for better performance. Since Spark 2.4, Spark converts ORC Hive tables by default, too. It means Spark uses its own ORC support by default instead of Hive SerDe. As an example, `CREATE TABLE t(id int) STORED AS ORC` would be handled with Hive SerDe in Spark 2.3, and in Spark 2.4, it would be converted into Spark's ORC data source table and ORC vectorization would be applied. To set `false` to `spark.sql.hive.convertMetastoreOrc` restores the previous behavior.
+ - Since Spark 2.4, handling of malformed rows in CSV files was changed. Previously, all column values of every row are parsed independently of its future usage. A row was considered as malformed if the CSV parser wasn't able to handle any column value in the row even if the value wasn't requested. Starting from version 2.4, only requested column values are parsed, and other values can be ignored. In such way, correct column values that were considered as malformed in previous Spark version only because of other malformed values become correct in Spark version 2.4.
--- End diff --
Shall we add some more examples? I guess now df.count() with dropmalformed give a different number too.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90473 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90473/testReport)** for PR 21296 at commit [`a4a0a54`](https://github.com/apache/spark/commit/a4a0a549156a15011c33c7877a35f244d75b7a4f).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21296
LGTM except a minor comment about the config name
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90752/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90555 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90555/testReport)** for PR 21296 at commit [`f90daa7`](https://github.com/apache/spark/commit/f90daa7ea33d119be978c27de10978c2d6281e25).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21296
thanks, merging to master!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on the issue:
https://github.com/apache/spark/pull/21296
Just for clarification, I'm okay.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/90473/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] Passing only required columns to the ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21296
**[Test build #90752 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/90752/testReport)** for PR 21296 at commit [`f89eeb7`](https://github.com/apache/spark/commit/f89eeb7f7ba86888ad3f7994577a4d4ebbf09197).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21296: [SPARK-24244][SQL] CSV column pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21296
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21296: [SPARK-24244][SQL] Passing only required columns ...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21296#discussion_r187762615
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala ---
@@ -73,11 +64,24 @@ class UnivocityParser(
// Each input token is placed in each output row's position by mapping these. In this case,
//
// output row - ["A", 2]
- private val valueConverters: Array[ValueConverter] =
- schema.map(f => makeConverter(f.name, f.dataType, f.nullable, options)).toArray
+ private val valueConverters: Array[ValueConverter] = {
+ requiredSchema.map(f => makeConverter(f.name, f.dataType, f.nullable, options)).toArray
+ }
- private val tokenIndexArr: Array[Int] = {
- requiredSchema.map(f => schema.indexOf(f)).toArray
+ private val tokenizer = {
+ val parserSetting = options.asParserSettings
+ if (requiredSchema.length < schema.length) {
+ val tokenIndexArr = requiredSchema.map(f => java.lang.Integer.valueOf(schema.indexOf(f)))
+ parserSetting.selectIndexes(tokenIndexArr: _*)
+ }
+ new CsvParser(parserSetting)
+ }
+
+ private val row = new GenericInternalRow(requiredSchema.length)
--- End diff --
Seems we don't need to move this down.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org