It executes two foreground statements in Confluent Cloud. The results of both statements are
* printed to the console.
*/
public class hello_table_api {
// All logic is defined in a main() method. It can run both in an IDE or CI/CD system.
public static void main(String[] args) {
// Set up connection properties to Confluent Cloud.
// Use the fromGlobalVariables() method if you assigned environment variables.
// EnvironmentSettings settings = ConfluentSettings.fromGlobalVariables();
// Use the fromArgs(args) method if you want to run with command-line arguments.
EnvironmentSettings settings = ConfluentSettings.fromArgs(args);
// Initialize the session context to get started.
TableEnvironment env = TableEnvironment.create(settings);
System.out.println("Running with printing...");
// The Table API centers on 'Table' objects, which help in defining data pipelines
// fluently. You can define pipelines fully programmatically.
Table table = env.fromValues("Hello world!");
// Also, You can define pipelines with embedded Flink SQL.
}
```
--------------------------------
### Build and Run Java JAR for Confluent Cloud
Source: https://docs.confluent.io/cloud/current/flink/get-started/quick-start-java-table-api.html
This section provides Maven and Java commands to build a JAR file from your Flink Table API project and then execute it. It includes example command-line arguments for configuring the connection to Confluent Cloud, such as cloud provider, region, API keys, and environment/compute pool IDs.
```bash
mvn clean package
```
```bash
java -jar target/flink-table-api-java-hello-world-1.0.jar \
--cloud aws \
--region us-east-1 \
--flink-api-key key \
--flink-api-secret secret \
--organization-id b0b21724-4586-4a07-b787-d0bb5aacbf87 \
--environment-id env-z3y2x1 \
--compute-pool-id lfcp-8m03rm
```
--------------------------------
### Sphinx Documentation Setup (JavaScript)
Source: https://docs.confluent.io/cloud/current/flink/concepts/dynamic-tables.html
Initializes Sphinx documentation theme options and enables sticky navigation. This code is typically found in a JavaScript file linked to the documentation.
```javascript
var DOCUMENTATION_OPTIONS = { URL_ROOT: '', VERSION: 'current', COLLAPSE_INDEX: false, FILE_SUFFIX: '.html', HAS_SOURCE: true };
jQuery(function () {
SphinxRtdTheme.StickyNav.enable();
});
```
--------------------------------
### Start Flink SQL Shell CLI
Source: https://docs.confluent.io/cloud/current/flink/get-started/quick-start-shell.html
Starts the Flink SQL shell, which connects to Confluent Cloud. This command requires the compute pool ID and environment ID. Statements submitted without a `--service-account` option run using the user's account.
```bash
confluent flink shell --compute-pool ${COMPUTE_POOL_ID} --environment ${ENV_ID}
```
--------------------------------
### Install and Use uv for Python Environment Management
Source: https://docs.confluent.io/cloud/current/flink/get-started/quick-start-python-table-api.html
Provides commands to install the uv package manager and create a virtual environment for Python projects. uv is used to manage Python versions and dependencies, ensuring compatibility for Flink Table API programs.
```bash
# Use one of the following commands to install uv.
curl -LsSf https://astral.sh/uv/install.sh | sh
# or
brew install uv
# or
pip install uv
# Create a new virtual environment.
uv venv --python 3.11
```
--------------------------------
### Inspect Example Stream with Flink SQL
Source: https://docs.confluent.io/cloud/current/flink/how-to-guides/mask-fields.html
This statement queries the 'customers' table in the 'examples.marketplace' database to display the data within the example stream. It is used to inspect the raw data before applying any transformations.
```sql
SELECT * FROM examples.marketplace.customers;
```
--------------------------------
### Flink SQL CREATE VIEW Usage Example
Source: https://docs.confluent.io/cloud/current/flink/reference/statements/create-view.html
An example demonstrating how to create a Flink SQL view named 'customer_orders' that calculates the total spent per customer from an 'orders' table. This view can then be queried like a regular table.
```sql
CREATE VIEW customer_orders AS
SELECT customer_id, SUM(price) AS total_spent
FROM `examples`.`marketplace`.`orders`
GROUP BY customer_id;
```
```sql
SELECT customer_id, total_spent
FROM customer_orders
WHERE total_spent > 1000;
```
--------------------------------
### Flink SQL EXPLAIN Basic Query Analysis Example
Source: https://docs.confluent.io/cloud/current/flink/reference/statements/explain.html
This example demonstrates the usage of the EXPLAIN statement to analyze a basic Flink SQL query that finds users who clicked but never placed an order. The output provides the physical plan and operator details for the query.
```sql
EXPLAIN
SELECT c.*
FROM `examples`.`marketplace`.`clicks` c
LEFT JOIN (
SELECT DISTINCT customer_id
FROM `examples`.`marketplace`.`orders`
) o ON c.user_id = o.customer_id
WHERE o.customer_id IS NULL;
```
--------------------------------
### Basic Flink SQL Queries
Source: https://docs.confluent.io/cloud/current/flink/reference/queries/overview.html
Examples of simple Flink SQL statements. The first demonstrates printing a string, and the second shows aggregation using COUNT and GROUP BY on a derived table.
```sql
SELECT 'Hello SQL';
```
```sql
SELECT Name, COUNT(*) AS Num
FROM
(VALUES ('Neo'), ('Trinity'), ('Morpheus'), ('Trinity')) AS NameTable(Name)
GROUP BY Name;
```
--------------------------------
### Avro to Flink SQL Type Mapping Example (JSON)
Source: https://docs.confluent.io/cloud/current/flink/reference/serialization.html
Provides an example of mapping Avro types to Flink SQL types, particularly for types not directly covered by the previous table. This is crucial for schema inference when consuming external Avro schemas.
```json
[
{
"name": "long",
"avro_logical_type": "time-micros",
"flink_sql_type": "BIGINT"
},
{
"name": "enum",
"avro_logical_type": null,
"flink_sql_type": "STRING"
},
{
"name": "union with null type (null + one other type)",
"avro_logical_type": null,
"flink_sql_type": "NULLABLE(type)"
},
{
"name": "union (other unions)",
"avro_logical_type": null,
"flink_sql_type": "ROW(type_name Type0, …)"
},
{
"name": "string (uuid)",
"avro_logical_type": null,
"flink_sql_type": "STRING"
},
{
"name": "fixed (duration)",
"avro_logical_type": null,
"flink_sql_type": "BINARY(size)"
}
]
```
--------------------------------
### Cross-Environment Query Example in Flink SQL
Source: https://docs.confluent.io/cloud/current/flink/overview.html
An example of a three-part name query in Flink SQL to access data across different environments. This demonstrates how to query data from another environment using a specified database and table. Ensure appropriate RBAC permissions are configured for data access.
```sql
SELECT * FROM `myEnvironment`.`myDatabase`.`myTable`;
```
--------------------------------
### Initialize Sphinx Documentation Theme
Source: https://docs.confluent.io/cloud/current/flink/operate-and-deploy/best-practices.html
Initializes the Sticky Navigation for the Sphinx documentation theme. This is a common setup script for projects using Sphinx for documentation generation.
```javascript
var DOCUMENTATION_OPTIONS = {
URL_ROOT: '',
VERSION: 'current',
COLLAPSE_INDEX: false,
FILE_SUFFIX: '.html',
HAS_SOURCE: true
};
jQuery(function () {
SphinxRtdTheme.StickyNav.enable();
});
```
--------------------------------
### Flink SQL: Override Scan Startup Mode
Source: https://docs.confluent.io/cloud/current/flink/reference/statements/hints.html
Example of overriding the 'scan.startup.mode' for a table in Flink SQL using dynamic options. This allows specifying how the query should start reading from a table, such as from the earliest or latest offset.
```sql
SELECT id, name
FROM table /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
```
```sql
SELECT *
FROM orders /*+ OPTIONS('scan.startup.mode'='latest-offset') */;
```
--------------------------------
### Initial Data Selection (SQL)
Source: https://docs.confluent.io/cloud/current/flink/how-to-guides/transform-topic.html
This SQL query selects the 'regionid' column from a source table within a specified environment, cluster, and dataset. It's an example of an initial data retrieval step before transformations.
```sql
SELECT `regionid`
FROM `your-env`.`your-cluster`.`users`;
```
--------------------------------
### After-Match Strategy: SKIP TO NEXT ROW
Source: https://docs.confluent.io/cloud/current/flink/reference/queries/match_recognize.html
This example describes the 'SKIP TO NEXT ROW' after-match strategy. It dictates that pattern matching restarts at the row immediately following the starting row of the completed match. This allows for overlapping matches.
```plaintext
AFTER MATCH SKIP TO NEXT ROW
```
--------------------------------
### Initialize Documentation Options (JavaScript)
Source: https://docs.confluent.io/cloud/current/flink/get-started/overview.html
Initializes configuration options for Sphinx documentation. This includes setting the root URL, version, and file suffix for the documentation site. It's a common pattern for Sphinx-generated documentation.
```javascript
var DOCUMENTATION_OPTIONS = {
URL_ROOT: '',
VERSION: 'current',
COLLAPSE_INDEX: false,
FILE_SUFFIX: '.html',
HAS_SOURCE: true
};
```
--------------------------------
### Start Flink Shell using Confluent CLI
Source: https://docs.confluent.io/cloud/current/flink/how-to-guides/create-udf.html
This command initiates the Flink shell environment using the Confluent CLI. It requires specifying the environment ID and compute pool ID for the Confluent Cloud setup where the Flink jobs will run.
```bash
confluent flink shell \
--environment ${ENV_ID} \
--compute-pool ${COMPUTE_POOL_ID}
```
--------------------------------
### Set Default Catalog (SQL)
Source: https://docs.confluent.io/cloud/current/flink/get-started/quick-start-shell.html
Sets the 'examples' catalog as the default for subsequent SQL operations. This simplifies table referencing by allowing three-part identifiers (catalog.database.table) to be shortened.
```SQL
USE CATALOG `examples`;
```
--------------------------------
### Get Current Date - SQL
Source: https://docs.confluent.io/cloud/current/flink/reference/functions/datetime-functions.html
The CURRENT_DATE function returns the current SQL date in the local time zone. In streaming mode, it's evaluated per record. In batch mode, it's evaluated once at query start, returning the same date for all rows.
```sql
SELECT CURRENT_DATE;
```
--------------------------------
### Get Current Timestamp - LOCALTIMESTAMP
Source: https://docs.confluent.io/cloud/current/flink/reference/functions/datetime-functions.html
The LOCALTIMESTAMP function retrieves the current SQL timestamp in the local time zone. Its return type is TIMESTAMP(3). In streaming mode, it's evaluated per record; in batch mode, it's evaluated once at query start.
```sql
SELECT LOCALTIMESTAMP;
```
--------------------------------
### Flink SQL CREATE TABLE Examples
Source: https://docs.confluent.io/cloud/current/flink/reference/sql-examples.html
Demonstrates creating Flink tables with different configurations in Confluent Cloud for Apache Flink. This includes minimal table creation with default properties for changelog mode, Schema Registry, distribution, and Kafka partitions.
```sql
CREATE TABLE t_minimal (
s STRING
);
```
--------------------------------
### Get Current Local Time
Source: https://docs.confluent.io/cloud/current/flink/reference/functions/datetime-functions.html
The LOCALTIME function returns the current SQL time in the local time zone. The return type is TIME(0). In streaming mode, it's evaluated per record; in batch mode, it's evaluated once at query start.
```sql
SELECT LOCALTIME;
-- returns the local machine time as "hh:mm:ss", for example:
-- 13:16:03
```
--------------------------------
### Window Deduplication SQL Query Example
Source: https://docs.confluent.io/cloud/current/flink/reference/queries/window-deduplication.html
This SQL query demonstrates how to remove duplicate rows within a 10-minute tumbling window, keeping the last record based on event time. It utilizes the ROW_NUMBER() function for partitioning by window start and end times, along with specified keys.
```sql
SELECT
column_list
FROM (
SELECT
column_list,
ROW_NUMBER() OVER (PARTITION BY window_start, window_end [, column_key1...]
ORDER BY time_attr [asc|desc]) AS rownum
FROM table_name) -- relation applied windowing TVF
WHERE (rownum = 1 | rownum <=1 | rownum < 2) [AND conditions]
```
--------------------------------
### Create and Populate Flink SQL Customers Table
Source: https://docs.confluent.io/cloud/current/flink/reference/example-data.html
This Flink SQL code creates a 'customers_source' table with a defined schema and then populates it by selecting all data from the 'examples.marketplace.customers' stream. The INSERT INTO statement runs continuously until stopped manually. It requires a Flink environment configured to access the 'examples' catalog.
```sql
CREATE TABLE customers_source (
customer_id INT,
name STRING,
address STRING,
postcode STRING,
city STRING,
email STRING,
PRIMARY KEY (customer_id) NOT ENFORCED
);
INSERT INTO customers_source(
customer_id,
name,
address,
postcode,
city,
email
)
SELECT * FROM examples.marketplace.customers;
```
--------------------------------
### SQL Query to Join Orders, Customers, and Products Tables
Source: https://docs.confluent.io/cloud/current/flink/reference/example-data.html
Demonstrates how to join the 'products', 'orders', and 'customers' data streams to display customer names, product names, and order prices. It uses standard SQL JOIN syntax.
```sql
SELECT
examples.marketplace.customers.name AS customer_name,
examples.marketplace.products.name AS product_name,
examples.marketplace.orders.price
FROM examples.marketplace.products
JOIN examples.marketplace.orders ON examples.marketplace.products.product_id = examples.marketplace.orders.product_id
JOIN examples.marketplace.customers ON examples.marketplace.customers.customer_id = examples.marketplace.orders.customer_id;
```
--------------------------------
### Initialize Sphinx Documentation Theme
Source: https://docs.confluent.io/cloud/current/flink/concepts/delivery-guarantees.html
Initializes the StickyNav functionality for the Sphinx Read the Docs theme. This is typically found in documentation build outputs.
```javascript
var DOCUMENTATION_OPTIONS = {
URL_ROOT: '',
VERSION: 'current',
COLLAPSE_INDEX: false,
FILE_SUFFIX: '.html',
HAS_SOURCE: true
};
jQuery(function () {
SphinxRtdTheme.StickyNav.enable();
});
```
--------------------------------
### ML_RECURSIVE_TEXT_SPLITTER SQL Example
Source: https://docs.confluent.io/cloud/current/flink/reference/functions/ml-preprocessing-functions.html
Provides an SQL example for the ML_RECURSIVE_TEXT_SPLITTER function, which divides text into smaller chunks recursively using specified separators. This example demonstrates custom separators and regex usage.
```sql
SELECT ML_RECURSIVE_TEXT_SPLITTER('Hello. world!', 0, 0, ARRAY['\[!\]','\[.\]'], TRUE, TRUE, TRUE, 'START');
```
--------------------------------
### Flink SQL Cascading Window Aggregation Example
Source: https://docs.confluent.io/cloud/current/flink/reference/queries/window-aggregation.html
This Flink SQL query demonstrates a cascading window aggregation. It first calculates a 5-minute tumbling window sum of points for each player and then performs a 10-minute tumbling window aggregation on the results, propagating the time attribute for subsequent operations. The inner window TVF's window start and end fields are aliased to prevent conflicts.
```sql
WITH fiveminutewindow AS (
SELECT window_start AS window_5mintumble_start, window_end as window_5mintumble_end, window_time AS rowtime, SUM(points) as `partial_sum`
FROM TABLE(
TUMBLE(TABLE `examples`.`marketplace`.`orders`, DESCRIPTOR($rowtime), INTERVAL '5' MINUTES))
GROUP BY player_id, window_start, window_end, window_time
)
SELECT window_start, window_end, SUM(partial_price) as total_price
FROM TABLE(
TUMBLE(TABLE fiveminutewindow, DESCRIPTOR($rowtime), INTERVAL '10' MINUTES))
GROUP BY window_start, window_end
```
--------------------------------
### Initialize Sphinx Documentation Theme
Source: https://docs.confluent.io/confluent-cli/current/command-reference/flink/connection/confluent_flink_connection_describe.html
Initializes the StickyNav functionality for the Read the Docs theme in Sphinx projects. This script is typically run after the DOM is ready to ensure smooth navigation.
```javascript
var DOCUMENTATION_OPTIONS = {
URL_ROOT: '',
VERSION: '4.41.0',
COLLAPSE_INDEX: false,
FILE_SUFFIX: '.html',
HAS_SOURCE: true
};
jQuery(function () {
SphinxRtdTheme.StickyNav.enable();
});
```
--------------------------------
### Protobuf Schema and SQL Table for Purchase and Pageview
Source: https://docs.confluent.io/cloud/current/flink/reference/sql-examples.html
Defines a Protobuf schema with separate 'Purchase' and 'Pageview' messages. SHOW CREATE TABLE then displays the equivalent SQL table structure, handling nested Protobuf messages.
```protobuf
syntax = "proto3";
message Purchase {
string item = 1;
double amount = 2;
string customer_id = 3;
}
message Pageview {
string url = 1;
bool is_special = 2;
string customer_id = 3;
}
```
```sql
CREATE TABLE `t` (
`key` VARBINARY(2147483647),
`Purchase` ROW<
`item` VARCHAR(2147483647) NOT NULL,
`amount` DOUBLE NOT NULL,
`customer_id` VARCHAR(2147483647) NOT NULL
>,
`Pageview` ROW<
`url` VARCHAR(2147483647) NOT NULL,
`is_special` BOOLEAN NOT NULL,
`customer_id` VARCHAR(2147483647) NOT NULL
>
)
```
--------------------------------
### ALTER MODEL Examples - Flink SQL
Source: https://docs.confluent.io/cloud/current/flink/reference/statements/alter-model.html
A collection of examples illustrating the practical usage of the ALTER MODEL statement in Flink SQL. These examples cover renaming a model, conditionally renaming a model if it exists, updating specific model version options, and resetting options to their default values.
```sql
-- Rename a model.
ALTER MODEL `my_model` RENAME TO `my_new_model`
-- Check for model existence and rename if it exists.
ALTER MODEL IF EXISTS `my_model` RENAME TO `my_new_model`
-- Change options for version 2.
ALTER MODEL `my_model$2` SET (
tag = 'prod',
description = "new_description"
);
-- Reset the tag option.
ALTER MODEL `my_model` RESET (tag)
```
--------------------------------
### Initialize Documentation Options (JavaScript)
Source: https://docs.confluent.io/confluent-cli/current/command-reference/flink/confluent_flink_shell.html
This snippet initializes global documentation options for a Sphinx-based documentation site. It sets the root URL, version, collapse state for the index, file suffix, and source availability. This is essential for the correct functioning of the documentation's navigation and features.
```javascript
var DOCUMENTATION_OPTIONS = {
URL_ROOT: '',
VERSION: '4.41.0',
COLLAPSE_INDEX: false,
FILE_SUFFIX: '.html',
HAS_SOURCE: true
};
```
--------------------------------
### Immutable Statement Property Example
Source: https://docs.confluent.io/cloud/current/flink/concepts/schema-statement-evolution.html
An example of an immutable statement property, specifically the SQL state time-to-live (TTL). This setting cannot be changed after the statement is created.
```sql
'sql.state-ttl' = '1h'
```
--------------------------------
### Immutable Query Example
Source: https://docs.confluent.io/cloud/current/flink/concepts/schema-statement-evolution.html
An example of an immutable query component of a statement. This SQL query defines the data retrieval and transformation logic without any mutable properties.
```sql
SELECT
v_orders.product,
to_minor_currency(v_orders.price),
customers.*
FROM orders
JOIN customers FOR SYSTEM TIME AS OF orders.$rowtime
ON v_orders.customer_id = customers.id;
```
--------------------------------
### Query Products Data Stream (SQL)
Source: https://docs.confluent.io/cloud/current/flink/reference/example-data.html
A SQL query to inspect the 'products' data stream. It selects all columns from the 'examples.marketplace.products' stream.
```sql
SELECT * FROM examples.marketplace.products;
```
--------------------------------
### SQL VAR_POP Function Example
Source: https://docs.confluent.io/cloud/current/flink/reference/functions/aggregate-functions.html
Calculates the population variance, which is the square of the population standard deviation, for an expression over all input rows. The example demonstrates its usage with a set of values.
```SQL
SELECT VAR_POP(my_values)
FROM (VALUES (0.5), (1.5), (2.2), (3.2)) AS my_values;
```
--------------------------------
### Set Default Database (SQL)
Source: https://docs.confluent.io/cloud/current/flink/get-started/quick-start-shell.html
Sets the 'marketplace' database as the default within the currently active catalog. This further simplifies queries by allowing two-part identifiers (database.table) or just table names if the catalog is also set.
```SQL
USE `marketplace`;
```
--------------------------------
### Maven Shade Plugin Configuration for JAR Packaging
Source: https://docs.confluent.io/cloud/current/flink/get-started/quick-start-java-table-api.html
This configuration snippet from a Maven pom.xml file defines the `maven-shade-plugin`. It is used to create an executable JAR, specifying transformers like `ManifestResourceTransformer` to include a main class entry.
```xml