You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "YannByron (via GitHub)" <gi...@apache.org> on 2023/09/18 08:18:15 UTC
[GitHub] [incubator-paimon] YannByron opened a new pull request, #2034: [spark] structured streaming docs
YannByron opened a new pull request, #2034:
URL: https://github.com/apache/incubator-paimon/pull/2034
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on a diff in pull request #2034: [spark] structured streaming docs
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on code in PR #2034:
URL: https://github.com/apache/incubator-paimon/pull/2034#discussion_r1330926700
##########
docs/content/engines/spark3.md:
##########
@@ -160,52 +158,244 @@ INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
create table my_table (
k int,
v string
-) USING paimon tblproperties (
+) USING paimon
+tblproperties (
'primary-key' = 'k'
) ;
-INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
```
{{< /tab >}}
{{< /tabs >}}
-**Step 4: Query Table with SQL**
+## Insert Table
+
+```sql
+INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
+```
+
+## Query Table
+
+{{< tabs "Query Paimon Table" >}}
+
+{{< tab "SQL" >}}
```sql
SELECT * FROM my_table;
+
/*
1 Hi
2 Hello
*/
```
-**Step 5: Update the Records**
+{{< /tab >}}
+
+{{< tab "DataFrame" >}}
+
+```scala
+val dataset = spark.read.format("paimon").load("file:/tmp/paimon/default.db/my_table")
+dataset.show()
+
+/*
++---+------+
+| k | v|
++---+------+
+| 1| Hi|
+| 2| Hello|
++---+------+
+*/
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
+
+## Update Table
+
+For now, Paimon does not support `UPDATE` syntax. But we can use `INSERT INTO` syntax instead for changelog tables.
```sql
INSERT INTO my_table VALUES (1, 'Hi Again'), (3, 'Test');
SELECT * FROM my_table;
+
/*
1 Hi Again
2 Hello
3 Test
*/
```
-**Step 6: Query Table with Scala API**
+## Streaming Write
-If you don't want to use Paimon catalog, you can also run `spark-shell` and query the table with Scala API.
+{{< hint info >}}
-```bash
-spark-shell ... --jars /path/to/paimon-spark-3.3-{{< version >}}.jar
+Paimon currently supports Spark 3+ for streaming write.
+
+Paimon Structured Streaming only supports the two `append` and `complete` modes.
+
+{{< /hint >}}
+
+```scala
+// Create a paimon table if not exists.
+spark.sql(s"""
+ |CREATE TABLE T (k INT, v STRING)
+ |TBLPROPERTIES ('primary-key'='a', 'write-mode'='change-log', 'bucket'='3')
+ |""".stripMargin)
+
+// Here we use MemoryStream to fake a streaming source.
+val inputData = MemoryStream[(Int, String)]
+val df = inputData.toDS().toDF("k", "v")
+
+// Streaming Write to paimon table.
+val stream = df
+ .writeStream
+ .outputMode("append")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .format("paimon")
+ .start("/path/to/paimon/sink/table")
```
+## Streaming Read
+
+{{< hint info >}}
+
+Paimon currently supports Spark 3.3+ for streaming read.
+
+{{< /hint >}}
+
+Paimon supports rich scan mode for streaming read. There is a list:
+<table class="configuration table table-bordered">
Review Comment:
what's the difference. See ` Spark Type Conversion` use the table style.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on pull request #2034: [spark] structured streaming docs
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on PR #2034:
URL: https://github.com/apache/incubator-paimon/pull/2034#issuecomment-1724909041
> And we should document Spark version for streaming read write.
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] structured streaming docs [incubator-paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #2034:
URL: https://github.com/apache/incubator-paimon/pull/2034
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on pull request #2034: [spark] structured streaming docs
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #2034:
URL: https://github.com/apache/incubator-paimon/pull/2034#issuecomment-1724948820
@YannByron, does the `Compatibility Matrix` of overview document need to sync?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] YannByron commented on pull request #2034: [spark] structured streaming docs
Posted by "YannByron (via GitHub)" <gi...@apache.org>.
YannByron commented on PR #2034:
URL: https://github.com/apache/incubator-paimon/pull/2034#issuecomment-1724909555
> Document for `write.merge-schema` too?
let open another pr for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #2034: [spark] structured streaming docs
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #2034:
URL: https://github.com/apache/incubator-paimon/pull/2034#discussion_r1329665017
##########
docs/content/engines/spark3.md:
##########
@@ -160,52 +158,244 @@ INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
create table my_table (
k int,
v string
-) USING paimon tblproperties (
+) USING paimon
+tblproperties (
'primary-key' = 'k'
) ;
-INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
```
{{< /tab >}}
{{< /tabs >}}
-**Step 4: Query Table with SQL**
+## Insert Table
+
+```sql
+INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
+```
+
+## Query Table
+
+{{< tabs "Query Paimon Table" >}}
+
+{{< tab "SQL" >}}
```sql
SELECT * FROM my_table;
+
/*
1 Hi
2 Hello
*/
```
-**Step 5: Update the Records**
+{{< /tab >}}
+
+{{< tab "DataFrame" >}}
+
+```scala
+val dataset = spark.read.format("paimon").load("file:/tmp/paimon/default.db/my_table")
+dataset.show()
+
+/*
++---+------+
+| k | v|
++---+------+
+| 1| Hi|
+| 2| Hello|
++---+------+
+*/
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
+
+## Update Table
+
+For now, Paimon does not support `UPDATE` syntax. But we can use `INSERT INTO` syntax instead for changelog tables.
```sql
INSERT INTO my_table VALUES (1, 'Hi Again'), (3, 'Test');
SELECT * FROM my_table;
+
/*
1 Hi Again
2 Hello
3 Test
*/
```
-**Step 6: Query Table with Scala API**
+## Streaming Write
-If you don't want to use Paimon catalog, you can also run `spark-shell` and query the table with Scala API.
+{{< hint info >}}
-```bash
-spark-shell ... --jars /path/to/paimon-spark-3.3-{{< version >}}.jar
+Paimon currently supports Spark 3+ for streaming write.
+
+Paimon Structured Streaming only supports the two `append` and `complete` modes.
+
+{{< /hint >}}
+
+```scala
+// Create a paimon table if not exists.
+spark.sql(s"""
+ |CREATE TABLE T (k INT, v STRING)
+ |TBLPROPERTIES ('primary-key'='a', 'write-mode'='change-log', 'bucket'='3')
+ |""".stripMargin)
+
+// Here we use MemoryStream to fake a streaming source.
+val inputData = MemoryStream[(Int, String)]
+val df = inputData.toDS().toDF("k", "v")
+
+// Streaming Write to paimon table.
+val stream = df
+ .writeStream
+ .outputMode("append")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .format("paimon")
+ .start("/path/to/paimon/sink/table")
```
+## Streaming Read
+
+{{< hint info >}}
+
+Paimon currently supports Spark 3.3+ for streaming read.
+
+{{< /hint >}}
+
+Paimon supports rich scan mode for streaming read. There is a list:
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Scan Mode</th>
+ <th class="text-left" style="width: 60%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>latest</h5></td>
+ <td>For streaming sources, continuously reads latest changes without producing a snapshot at the beginning. </td>
+ </tr>
+ <tr>
+ <td><h5>latest-full</h5></td>
+ <td>For streaming sources, produces the latest snapshot on the table upon first startup, and continue to read the latest changes.</td>
+ </tr>
+ <tr>
+ <td><h5>from-timestamp</h5></td>
+ <td>For streaming sources, continuously reads changes starting from timestamp specified by "scan.timestamp-millis", without producing a snapshot at the beginning. </td>
+ </tr>
+ <tr>
+ <td><h5>from-snapshot</h5></td>
+ <td>For streaming sources, continuously reads changes starting from snapshot specified by "scan.snapshot-id", without producing a snapshot at the beginning. </td>
+ </tr>
+ <tr>
+ <td><h5>from-snapshot-full</h5></td>
+ <td>For streaming sources, produces from snapshot specified by "scan.snapshot-id" on the table upon first startup, and continuously reads changes.</td>
+ </tr>
+ <tr>
+ <td><h5>default</h5></td>
+ <td>It is equivalent to from-snapshot if "scan.snapshot-id" is specified. It is equivalent to from-timestamp if "timestamp-millis" is specified. Or, It is equivalent to latest-full.</td>
+ </tr>
+ </tbody>
+</table>
+
+A simple example with default scan mode:
+
```scala
-val dataset = spark.read.format("paimon").load("file:/tmp/paimon/default.db/my_table")
-dataset.createOrReplaceTempView("my_table")
-spark.sql("SELECT * FROM my_table").show()
+// no any scan-related configs are provided, that will use latest-full scan mode.
+val query = spark.readStream
+ .format("paimon")
+ .load("/path/to/paimon/source/table")
+ .writeStream
+ .format("console")
+ .start()
+```
+
+Paimon Structured Streaming also supports a variety of streaming read modes, it can support many triggers and many read limits.
+
+These read limits are supported:
+
+<table class="configuration table table-bordered">
Review Comment:
Ditto.
##########
docs/content/engines/spark3.md:
##########
@@ -160,52 +158,244 @@ INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
create table my_table (
k int,
v string
-) USING paimon tblproperties (
+) USING paimon
+tblproperties (
'primary-key' = 'k'
) ;
-INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
```
{{< /tab >}}
{{< /tabs >}}
-**Step 4: Query Table with SQL**
+## Insert Table
+
+```sql
+INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
+```
+
+## Query Table
+
+{{< tabs "Query Paimon Table" >}}
+
+{{< tab "SQL" >}}
```sql
SELECT * FROM my_table;
+
/*
1 Hi
2 Hello
*/
```
-**Step 5: Update the Records**
+{{< /tab >}}
+
+{{< tab "DataFrame" >}}
+
+```scala
+val dataset = spark.read.format("paimon").load("file:/tmp/paimon/default.db/my_table")
+dataset.show()
+
+/*
++---+------+
+| k | v|
++---+------+
+| 1| Hi|
+| 2| Hello|
++---+------+
+*/
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
+
+## Update Table
+
+For now, Paimon does not support `UPDATE` syntax. But we can use `INSERT INTO` syntax instead for changelog tables.
```sql
INSERT INTO my_table VALUES (1, 'Hi Again'), (3, 'Test');
SELECT * FROM my_table;
+
/*
1 Hi Again
2 Hello
3 Test
*/
```
-**Step 6: Query Table with Scala API**
+## Streaming Write
-If you don't want to use Paimon catalog, you can also run `spark-shell` and query the table with Scala API.
+{{< hint info >}}
-```bash
-spark-shell ... --jars /path/to/paimon-spark-3.3-{{< version >}}.jar
+Paimon currently supports Spark 3+ for streaming write.
+
+Paimon Structured Streaming only supports the two `append` and `complete` modes.
+
+{{< /hint >}}
+
+```scala
+// Create a paimon table if not exists.
+spark.sql(s"""
+ |CREATE TABLE T (k INT, v STRING)
+ |TBLPROPERTIES ('primary-key'='a', 'write-mode'='change-log', 'bucket'='3')
+ |""".stripMargin)
+
+// Here we use MemoryStream to fake a streaming source.
+val inputData = MemoryStream[(Int, String)]
+val df = inputData.toDS().toDF("k", "v")
+
+// Streaming Write to paimon table.
+val stream = df
+ .writeStream
+ .outputMode("append")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .format("paimon")
+ .start("/path/to/paimon/sink/table")
```
+## Streaming Read
+
+{{< hint info >}}
+
+Paimon currently supports Spark 3.3+ for streaming read.
+
+{{< /hint >}}
+
+Paimon supports rich scan mode for streaming read. There is a list:
+<table class="configuration table table-bordered">
Review Comment:
Could we use the following style for configuration?
```
| Scan Mode | Description |
|-------------|--------------------------------------------------------|
| latest | For streaming sources, continuously reads latest changes without producing a snapshot at the beginning.|
```
##########
docs/content/engines/spark3.md:
##########
@@ -160,52 +158,244 @@ INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
create table my_table (
k int,
v string
-) USING paimon tblproperties (
+) USING paimon
+tblproperties (
'primary-key' = 'k'
) ;
-INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
```
{{< /tab >}}
{{< /tabs >}}
-**Step 4: Query Table with SQL**
+## Insert Table
+
+```sql
+INSERT INTO my_table VALUES (1, 'Hi'), (2, 'Hello');
+```
+
+## Query Table
+
+{{< tabs "Query Paimon Table" >}}
+
+{{< tab "SQL" >}}
```sql
SELECT * FROM my_table;
+
/*
1 Hi
2 Hello
*/
```
-**Step 5: Update the Records**
+{{< /tab >}}
+
+{{< tab "DataFrame" >}}
+
+```scala
+val dataset = spark.read.format("paimon").load("file:/tmp/paimon/default.db/my_table")
+dataset.show()
+
+/*
++---+------+
+| k | v|
++---+------+
+| 1| Hi|
+| 2| Hello|
++---+------+
+*/
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
+
+## Update Table
+
+For now, Paimon does not support `UPDATE` syntax. But we can use `INSERT INTO` syntax instead for changelog tables.
```sql
INSERT INTO my_table VALUES (1, 'Hi Again'), (3, 'Test');
SELECT * FROM my_table;
+
/*
1 Hi Again
2 Hello
3 Test
*/
```
-**Step 6: Query Table with Scala API**
+## Streaming Write
-If you don't want to use Paimon catalog, you can also run `spark-shell` and query the table with Scala API.
+{{< hint info >}}
-```bash
-spark-shell ... --jars /path/to/paimon-spark-3.3-{{< version >}}.jar
+Paimon currently supports Spark 3+ for streaming write.
+
+Paimon Structured Streaming only supports the two `append` and `complete` modes.
+
+{{< /hint >}}
+
+```scala
+// Create a paimon table if not exists.
+spark.sql(s"""
+ |CREATE TABLE T (k INT, v STRING)
+ |TBLPROPERTIES ('primary-key'='a', 'write-mode'='change-log', 'bucket'='3')
+ |""".stripMargin)
+
+// Here we use MemoryStream to fake a streaming source.
+val inputData = MemoryStream[(Int, String)]
+val df = inputData.toDS().toDF("k", "v")
+
+// Streaming Write to paimon table.
+val stream = df
+ .writeStream
+ .outputMode("append")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .format("paimon")
+ .start("/path/to/paimon/sink/table")
```
+## Streaming Read
+
+{{< hint info >}}
+
+Paimon currently supports Spark 3.3+ for streaming read.
+
+{{< /hint >}}
+
+Paimon supports rich scan mode for streaming read. There is a list:
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Scan Mode</th>
+ <th class="text-left" style="width: 60%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>latest</h5></td>
+ <td>For streaming sources, continuously reads latest changes without producing a snapshot at the beginning. </td>
+ </tr>
+ <tr>
+ <td><h5>latest-full</h5></td>
+ <td>For streaming sources, produces the latest snapshot on the table upon first startup, and continue to read the latest changes.</td>
+ </tr>
+ <tr>
+ <td><h5>from-timestamp</h5></td>
+ <td>For streaming sources, continuously reads changes starting from timestamp specified by "scan.timestamp-millis", without producing a snapshot at the beginning. </td>
+ </tr>
+ <tr>
+ <td><h5>from-snapshot</h5></td>
+ <td>For streaming sources, continuously reads changes starting from snapshot specified by "scan.snapshot-id", without producing a snapshot at the beginning. </td>
+ </tr>
+ <tr>
+ <td><h5>from-snapshot-full</h5></td>
+ <td>For streaming sources, produces from snapshot specified by "scan.snapshot-id" on the table upon first startup, and continuously reads changes.</td>
+ </tr>
+ <tr>
+ <td><h5>default</h5></td>
+ <td>It is equivalent to from-snapshot if "scan.snapshot-id" is specified. It is equivalent to from-timestamp if "timestamp-millis" is specified. Or, It is equivalent to latest-full.</td>
+ </tr>
+ </tbody>
+</table>
+
+A simple example with default scan mode:
+
```scala
-val dataset = spark.read.format("paimon").load("file:/tmp/paimon/default.db/my_table")
-dataset.createOrReplaceTempView("my_table")
-spark.sql("SELECT * FROM my_table").show()
+// no any scan-related configs are provided, that will use latest-full scan mode.
+val query = spark.readStream
+ .format("paimon")
+ .load("/path/to/paimon/source/table")
+ .writeStream
+ .format("console")
+ .start()
+```
+
+Paimon Structured Streaming also supports a variety of streaming read modes, it can support many triggers and many read limits.
+
+These read limits are supported:
+
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>read.stream.maxFilesPerTrigger</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The maximum number of files returned in a single batch.</td>
+ </tr>
+ <tr>
+ <td><h5>read.stream.maxBytesPerTrigger</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Long</td>
+ <td>The maximum number of bytes returned in a single batch.</td>
+ </tr>
+ <tr>
+ <td><h5>read.stream.maxRowsPerTrigger</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Long</td>
+ <td>The maximum number of rows returned in a single batch.</td>
+ </tr>
+ <tr>
+ <td><h5>read.stream.minRowsPerTrigger</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Long</td>
+ <td>The minimum number of rows returned in a single batch, which used to create MinRowsReadLimit with read.stream.maxTriggerDelayMs together.</td>
+ </tr>
+ <tr>
+ <td><h5>read.stream.maxTriggerDelayMs</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Long</td>
+ <td>The maximum delay between two adjacent batches, which used to create MinRowsReadLimit with read.stream.minRowsPerTrigger together.</td>
+ </tr>
+ </tbody>
+</table>
+
+**Example: One**
Review Comment:
Could this add some simple description to describe what this example shows?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org