chore: convert splitting and cumulating windows Flink SQL tutorials to use INSERT INTO (#1644)

davetroiano · web-flow · commit 7c7da4c9a79f · 2023-10-20T12:04:56.000-04:00
diff --git a/_data/harnesses/cumulating-windows/flinksql.yml b/_data/harnesses/cumulating-windows/flinksql.yml
@@ -55,6 +55,10 @@ dev:
               render:
                 file: tutorials/cumulating-windows/flinksql/markup/dev/create-revenue-per-hour-cumulating.adoc
 
+            - file: tutorial-steps/dev/populate-revenue-per-hour-cumulating.sql
+              render:
+                file: tutorials/cumulating-windows/flinksql/markup/dev/populate-revenue-per-hour-cumulating.adoc
+
           stdout:
             directory: tutorial-steps/dev/outputs
 
@@ -114,6 +118,11 @@ test:
           render:
             file: tutorials/cumulating-windows/flinksql/markup/test/create-resource-create-revenue-per-hour-cumulating.sql.template.adoc
 
+        - action: make_file
+          file: src/test/resources/populate-revenue-per-hour-cumulating.sql
+          render:
+            file: tutorials/cumulating-windows/flinksql/markup/test/create-resource-populate-revenue-per-hour-cumulating.sql.adoc
+
         - action: make_file
           file: src/test/resources/query-revenue-per-hour-cumulating.sql
           render:
diff --git a/_data/harnesses/splitting/flinksql.yml b/_data/harnesses/splitting/flinksql.yml
@@ -59,14 +59,26 @@ dev:
               render:
                 file: tutorials/splitting/flinksql/markup/dev/create-acting-events-drama.adoc
 
+            - file: tutorial-steps/dev/populate-acting-events-drama.sql
+              render:
+                file: tutorials/splitting/flinksql/markup/dev/populate-acting-events-drama.adoc
+
             - file: tutorial-steps/dev/create-acting-events-fantasy.sql
               render:
                 file: tutorials/splitting/flinksql/markup/dev/create-acting-events-fantasy.adoc
 
+            - file: tutorial-steps/dev/populate-acting-events-fantasy.sql
+              render:
+                file: tutorials/splitting/flinksql/markup/dev/populate-acting-events-fantasy.adoc
+
             - file: tutorial-steps/dev/create-acting-events-other.sql
               render:
                 file: tutorials/splitting/flinksql/markup/dev/create-acting-events-other.adoc
 
+            - file: tutorial-steps/dev/populate-acting-events-other.sql
+              render:
+                file: tutorials/splitting/flinksql/markup/dev/populate-acting-events-other.adoc
+
           stdout:
             directory: tutorial-steps/dev/outputs
 
@@ -120,6 +132,11 @@ test:
           render:
             file: tutorials/splitting/flinksql/markup/test/create-resource-create-acting-events-drama.sql.template.adoc
 
+        - action: make_file
+          file: src/test/resources/populate-acting-events-drama.sql
+          render:
+            file: tutorials/splitting/flinksql/markup/test/create-resource-populate-acting-events-drama.sql.adoc
+
         - action: make_file
           file: src/test/resources/query-acting-events-drama.sql
           render:
diff --git a/_includes/tutorials/cumulating-windows/flinksql/code/settings.gradle b/_includes/tutorials/cumulating-windows/flinksql/code/settings.gradle
@@ -7,4 +7,4 @@
  * in the user manual at https://docs.gradle.org/6.7.1/userguide/multi_project_builds.html
  */
 
-rootProject.name = 'aggregating-count-flinksql'
+rootProject.name = 'cumulating-windows-flinksql'
diff --git a/_includes/tutorials/cumulating-windows/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlCumulatingWindowTest.java b/_includes/tutorials/cumulating-windows/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlCumulatingWindowTest.java
@@ -17,6 +17,8 @@ public void testCumulatingWindows() throws Exception {
         Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
     streamTableEnv.executeSql(getResourceFileContents("populate-orders.sql"));
     streamTableEnv.executeSql(getResourceFileContents("create-revenue-per-hour-cumulating.sql.template",
+        Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
+    streamTableEnv.executeSql(getResourceFileContents("populate-revenue-per-hour-cumulating.sql",
         Optional.of(kafkaPort), Optional.of(schemaRegistryPort)));
 
     TableResult tableResult = streamTableEnv.executeSql(getResourceFileContents("query-revenue-per-hour-cumulating.sql"));
diff --git a/_includes/tutorials/cumulating-windows/flinksql/code/src/test/resources/create-revenue-per-hour-cumulating.sql.template b/_includes/tutorials/cumulating-windows/flinksql/code/src/test/resources/create-revenue-per-hour-cumulating.sql.template
@@ -1,5 +1,8 @@
-CREATE TABLE revenue_per_hour_cumulating
-WITH (
+CREATE TABLE revenue_per_hour_cumulating (
+    revenue DOUBLE,
+    window_start TIMESTAMP(3),
+    window_end TIMESTAMP(3)
+) WITH (
     'connector' = 'kafka',
     'topic' = 'revenue-per-hour-cumulating',
     'properties.bootstrap.servers' = 'localhost:KAFKA_PORT',
@@ -10,9 +13,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://localhost:SCHEMA_REGISTRY_PORT',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT ROUND(SUM(quantity * unit_price), 2) AS revenue,
-        window_start,
-        window_end
-    FROM TABLE(CUMULATE(TABLE orders, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
-    GROUP BY window_start, window_end;
+);
diff --git a/_includes/tutorials/cumulating-windows/flinksql/code/src/test/resources/populate-revenue-per-hour-cumulating.sql b/_includes/tutorials/cumulating-windows/flinksql/code/src/test/resources/populate-revenue-per-hour-cumulating.sql
@@ -0,0 +1,6 @@
+INSERT INTO revenue_per_hour_cumulating
+    SELECT ROUND(SUM(quantity * unit_price), 2) AS revenue,
+        window_start,
+        window_end
+    FROM TABLE(CUMULATE(TABLE orders, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
+    GROUP BY window_start, window_end;
diff --git a/_includes/tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/create-revenue-per-hour-cumulating.sql b/_includes/tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/create-revenue-per-hour-cumulating.sql
@@ -1,5 +1,8 @@
-CREATE TABLE revenue_per_hour_cumulating
-WITH (
+CREATE TABLE revenue_per_hour_cumulating (
+    revenue DOUBLE,
+    window_start TIMESTAMP(3),
+    window_end TIMESTAMP(3)
+) WITH (
     'connector' = 'kafka',
     'topic' = 'revenue-per-hour-cumulating',
     'properties.bootstrap.servers' = 'broker:9092',
@@ -10,9 +13,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://schema-registry:8081',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT ROUND(SUM(quantity * unit_price), 2) AS revenue,
-        window_start,
-        window_end
-    FROM TABLE(CUMULATE(TABLE orders, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
-    GROUP BY window_start, window_end;
+);
diff --git a/_includes/tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/expected-revenue-per-hour-cumulating.log b/_includes/tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/expected-revenue-per-hour-cumulating.log
@@ -1,6 +1,6 @@
-{"revenue":{"double":13.96},"window_start":1673748000000,"window_end":1673748300000}
-{"revenue":{"double":33.94},"window_start":1673748000000,"window_end":1673748600000}
-{"revenue":{"double":9.99},"window_start":1673748600000,"window_end":1673749200000}
-{"revenue":{"double":26.67},"window_start":1673749200000,"window_end":1673749500000}
-{"revenue":{"double":61.57},"window_start":1673749200000,"window_end":1673749800000}
+{"revenue":{"double":13.96},"window_start":{"long":1673748000000},"window_end":{"long":1673748300000}}
+{"revenue":{"double":33.94},"window_start":{"long":1673748000000},"window_end":{"long":1673748600000}}
+{"revenue":{"double":9.99},"window_start":{"long":1673748600000},"window_end":{"long":1673749200000}}
+{"revenue":{"double":26.67},"window_start":{"long":1673749200000},"window_end":{"long":1673749500000}}
+{"revenue":{"double":61.57},"window_start":{"long":1673749200000},"window_end":{"long":1673749800000}}
 Processed a total of 5 messages
diff --git a/_includes/tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/populate-revenue-per-hour-cumulating.sql b/_includes/tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/populate-revenue-per-hour-cumulating.sql
@@ -0,0 +1,6 @@
+INSERT INTO revenue_per_hour_cumulating
+    SELECT ROUND(SUM(quantity * unit_price), 2) AS revenue,
+        window_start,
+        window_end
+    FROM TABLE(CUMULATE(TABLE orders, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
+    GROUP BY window_start, window_end;
diff --git a/_includes/tutorials/cumulating-windows/flinksql/markup/dev/create-revenue-per-hour-cumulating.adoc b/_includes/tutorials/cumulating-windows/flinksql/markup/dev/create-revenue-per-hour-cumulating.adoc
@@ -1,4 +1,4 @@
-Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE AS SELECT` statement. Go ahead and execute the following statement in your Flink SQL session:
+Since the output of our transient query looks right, the next step is to make it persistent with the following statements. Go ahead and execute them in your Flink SQL session:
 
 +++++
 <pre class="snippet"><code class="sql">{% include_raw tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/create-revenue-per-hour-cumulating.sql %}</code></pre>
diff --git a/_includes/tutorials/cumulating-windows/flinksql/markup/dev/populate-revenue-per-hour-cumulating.adoc b/_includes/tutorials/cumulating-windows/flinksql/markup/dev/populate-revenue-per-hour-cumulating.adoc
@@ -0,0 +1,3 @@
++++++
+<pre class="snippet"><code class="sql">{% include_raw tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/populate-revenue-per-hour-cumulating.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/cumulating-windows/flinksql/markup/test/create-resource-populate-revenue-per-hour-cumulating.sql.adoc b/_includes/tutorials/cumulating-windows/flinksql/markup/test/create-resource-populate-revenue-per-hour-cumulating.sql.adoc
@@ -0,0 +1,5 @@
+Next, create the following file at `src/test/resources/populate-revenue-per-hour-cumulating.sql`.
+
++++++
+<pre class="snippet"><code class="groovy">{% include_raw tutorials/cumulating-windows/flinksql/code/src/test/resources/populate-revenue-per-hour-cumulating.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/splitting/flinksql/code/settings.gradle b/_includes/tutorials/splitting/flinksql/code/settings.gradle
@@ -7,4 +7,4 @@
  * in the user manual at https://docs.gradle.org/6.7.1/userguide/multi_project_builds.html
  */
 
-rootProject.name = 'aggregating-count-flinksql'
+rootProject.name = 'splitting-flinksql'
diff --git a/_includes/tutorials/splitting/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlSplitStreamTest.java b/_includes/tutorials/splitting/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlSplitStreamTest.java
@@ -17,7 +17,8 @@ public void simpleSelect() throws Exception {
         Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
     streamTableEnv.executeSql(getResourceFileContents("populate-acting-events.sql"));
     streamTableEnv.executeSql(getResourceFileContents("create-acting-events-drama.sql.template",
-        Optional.of(kafkaPort), Optional.of(schemaRegistryPort)));
+        Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
+    streamTableEnv.executeSql(getResourceFileContents("populate-acting-events-drama.sql"));
 
     // execute query on result table that should have drama acting events
     TableResult tableResult = streamTableEnv.executeSql(getResourceFileContents("query-acting-events-drama.sql"));
diff --git a/_includes/tutorials/splitting/flinksql/code/src/test/resources/create-acting-events-drama.sql.template b/_includes/tutorials/splitting/flinksql/code/src/test/resources/create-acting-events-drama.sql.template
@@ -1,5 +1,7 @@
-CREATE TABLE acting_events_drama
-WITH (
+CREATE TABLE acting_events_drama (
+    name  STRING,
+    title STRING
+) WITH (
     'connector' = 'kafka',
     'topic' = 'acting-events-drama',
     'properties.bootstrap.servers' = 'localhost:KAFKA_PORT',
@@ -10,7 +12,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://localhost:SCHEMA_REGISTRY_PORT',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT name, title
-    FROM acting_events
-    WHERE genre = 'drama';
+);
diff --git a/_includes/tutorials/splitting/flinksql/code/src/test/resources/populate-acting-events-drama.sql b/_includes/tutorials/splitting/flinksql/code/src/test/resources/populate-acting-events-drama.sql
@@ -0,0 +1,4 @@
+INSERT INTO acting_events_drama
+    SELECT name, title
+    FROM acting_events
+    WHERE genre = 'drama';
diff --git a/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/create-acting-events-drama.sql b/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/create-acting-events-drama.sql
@@ -1,5 +1,7 @@
-CREATE TABLE acting_events_drama
-WITH (
+CREATE TABLE acting_events_drama (
+    name  STRING,
+    title STRING
+) WITH (
     'connector' = 'kafka',
     'topic' = 'acting-events-drama',
     'properties.bootstrap.servers' = 'broker:9092',
@@ -10,7 +12,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://schema-registry:8081',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT name, title
-    FROM acting_events
-    WHERE genre = 'drama';
+);
diff --git a/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/create-acting-events-fantasy.sql b/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/create-acting-events-fantasy.sql
@@ -1,5 +1,8 @@
-CREATE TABLE acting_events_fantasy
-WITH (
+CREATE TABLE acting_events_fantasy (
+    name  STRING,
+    title STRING
+)
+    WITH (
     'connector' = 'kafka',
     'topic' = 'acting-events-fantasy',
     'properties.bootstrap.servers' = 'broker:9092',
@@ -10,7 +13,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://schema-registry:8081',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT name, title
-    FROM acting_events
-    WHERE genre = 'fantasy';
+);
diff --git a/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/create-acting-events-other.sql b/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/create-acting-events-other.sql
@@ -1,5 +1,7 @@
-CREATE TABLE acting_events_other
-WITH (
+CREATE TABLE acting_events_other (
+    name  STRING,
+    title STRING
+) WITH (
     'connector' = 'kafka',
     'topic' = 'acting-events-other',
     'properties.bootstrap.servers' = 'broker:9092',
@@ -10,7 +12,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://schema-registry:8081',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT name, title
-    FROM acting_events
-    WHERE genre <> 'drama' AND genre <> 'fantasy';
+);
diff --git a/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-drama.sql b/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-drama.sql
@@ -0,0 +1,4 @@
+INSERT INTO acting_events_drama
+    SELECT name, title
+    FROM acting_events
+    WHERE genre = 'drama';
diff --git a/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-fantasy.sql b/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-fantasy.sql
@@ -0,0 +1,4 @@
+INSERT INTO acting_events_fantasy
+    SELECT name, title
+    FROM acting_events
+    WHERE genre = 'fantasy';
diff --git a/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-other.sql b/_includes/tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-other.sql
@@ -0,0 +1,4 @@
+INSERT INTO acting_events_other
+    SELECT name, title
+    FROM acting_events
+    WHERE genre <> 'drama' AND genre <> 'fantasy';
diff --git a/_includes/tutorials/splitting/flinksql/markup/dev/create-acting-events-drama.adoc b/_includes/tutorials/splitting/flinksql/markup/dev/create-acting-events-drama.adoc
@@ -1,4 +1,4 @@
-Since the output of our transient queries looks right, the next step is to make the queries persistent with the following `CREATE TABLE AS SELECT` statements. Go ahead and run the following three commands in your Flink SQL session:
+Since the output of our transient queries looks right, the next step is to make the queries persistent with the following statements. Go ahead and run the following six commands in your Flink SQL session:
 
 +++++
 <pre class="snippet"><code class="sql">{% include_raw tutorials/splitting/flinksql/code/tutorial-steps/dev/create-acting-events-drama.sql %}</code></pre>
diff --git a/_includes/tutorials/splitting/flinksql/markup/dev/populate-acting-events-drama.adoc b/_includes/tutorials/splitting/flinksql/markup/dev/populate-acting-events-drama.adoc
@@ -0,0 +1,3 @@
++++++
+<pre class="snippet"><code class="sql">{% include_raw tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-drama.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/splitting/flinksql/markup/dev/populate-acting-events-fantasy.adoc b/_includes/tutorials/splitting/flinksql/markup/dev/populate-acting-events-fantasy.adoc
@@ -0,0 +1,3 @@
++++++
+<pre class="snippet"><code class="sql">{% include_raw tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-fantasy.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/splitting/flinksql/markup/dev/populate-acting-events-other.adoc b/_includes/tutorials/splitting/flinksql/markup/dev/populate-acting-events-other.adoc
@@ -0,0 +1,3 @@
++++++
+<pre class="snippet"><code class="sql">{% include_raw tutorials/splitting/flinksql/code/tutorial-steps/dev/populate-acting-events-other.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/splitting/flinksql/markup/test/create-resource-populate-acting-events-drama.sql.adoc b/_includes/tutorials/splitting/flinksql/markup/test/create-resource-populate-acting-events-drama.sql.adoc
@@ -0,0 +1,4 @@
+Next create the following file at `src/test/resources/populate-acting-events-drama.sql`.
++++++
+<pre class="snippet"><code class="groovy">{% include_raw tutorials/splitting/flinksql/code/src/test/resources/populate-acting-events-drama.sql %}</code></pre>
++++++

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		-Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE AS SELECT` statement. Go ahead and execute the following statement in your Flink SQL session:
	`1`	`+Since the output of our transient query looks right, the next step is to make it persistent with the following statements. Go ahead and execute them in your Flink SQL session:`
`2`	`2`
`3`	`3`	`+++++`
`4`	`4`	`<pre class="snippet"><code class="sql">{% include_raw tutorials/cumulating-windows/flinksql/code/tutorial-steps/dev/create-revenue-per-hour-cumulating.sql %}</code></pre>`