redsk commented on a change in pull request #26153: [SPARK-29500][SQL][SS]
Support partition column when writing to Kafka
URL: https://github.com/apache/spark/pull/26153#discussion_r336456840
##########
File path:
external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSinkSuite.scala
##########
@@ -404,20 +420,85 @@ abstract class KafkaSinkBatchSuiteBase extends
KafkaSinkSuiteBase {
.save()
checkAnswer(
createKafkaReader(topic, includeHeaders = true).selectExpr(
- "CAST(value as STRING) value", "headers"
+ "CAST(value as STRING) value", "headers", "partition"
),
- Row("1", Seq(Row("a", "b".getBytes(UTF_8)))) ::
- Row("2", Seq(Row("c", "d".getBytes(UTF_8)), Row("e",
"f".getBytes(UTF_8)))) ::
- Row("3", Seq(Row("g", "h".getBytes(UTF_8)), Row("g",
"i".getBytes(UTF_8)))) ::
- Row("4", null) ::
+ Row("1", Seq(Row("a", "b".getBytes(UTF_8))), 0) ::
+ Row("2", Seq(Row("c", "d".getBytes(UTF_8)), Row("e",
"f".getBytes(UTF_8))), 1) ::
+ Row("3", Seq(Row("g", "h".getBytes(UTF_8)), Row("g",
"i".getBytes(UTF_8))), 2) ::
+ Row("4", null, 3) ::
Row("5", Seq(
Row("j", "k".getBytes(UTF_8)),
Row("j", "l".getBytes(UTF_8)),
- Row("m", "n".getBytes(UTF_8)))) ::
+ Row("m", "n".getBytes(UTF_8))), 0) ::
Nil
)
}
+ test("batch - partition column vs default Kafka partitioner") {
Review comment:
- "Not sure why 100 partitions necessary?": I just wanted to avoid the test
to succeed by chance. I admit it's overkill. I'll reduce it to 10.
- "Don't we need the following tests?". See my other comment.
- "`.collect().toList.head` maybe not enough because it would be good to
make sure the data is in one partition." `.collect()` docstring says "Returns
an array that contains all rows in this Dataset. Running collect requires
moving all the data into the application's driver process, and doing so on a
very large dataset can crash the driver process with OutOfMemoryError." Does
`collect()` work on a per-partition base? If so maybe I could use
`.coalesce(1)` before calling `collect()`.
- "I think this test can be formed more simple." Ok it makes sense. I'll
simplify it.
- "There are a couple of copy-pastes" Do you refer to the ```df.write
.format("kafka")
.option("kafka.bootstrap.servers", testUtils.brokerAddress)
.option("topic", topic)
.mode("append")
.save()``` blocks? In that case, they are also all over the file.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
