<a id="flink-sql-run-snapshot-query"></a>

# Run a Snapshot Query with Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® supports [snapshot queries](../concepts/snapshot-queries.md#flink-sql-snapshot-queries)
that read data from a table at a specific point in time. In contrast with a
streaming query, which runs continuously and returns results incrementally,
a snapshot query runs, returns results, and then exits.

This guide shows how to run a snapshot query on a Flink table.

- [Step 1: Create an example data stream](#flink-sql-run-snapshot-query-create-data-stream)
- [Step 2: Run a snapshot query on the topic (Flink workspace)](#flink-sql-run-snapshot-query-query-topic)
- [Step 3: Set the snapshot mode in SQL (alternative)](#flink-sql-run-snapshot-query-set-snapshot-mode)

## Prerequisites

You need the following prerequisites to use Flink in Confluent Cloud Console.

- Access to Confluent Cloud.

<a id="flink-sql-run-snapshot-query-create-data-stream"></a>

## Step 1: Create an example data stream

In this step, you create a Datagen source connector that produces a stream of
data.

If you have a topic with data, you can skip this step and proceed to
[Step 2: Run a snapshot query on the topic (Flink workspace)](#flink-sql-run-snapshot-query-query-topic).

1. In the Confluent Cloud UI, go to the **Environments** page.
2. Select the environment where you want to create the connector.
3. In the **Overview** page, click the cluster that you want to use.
4. In the navigation menu, click **Connectors**.
5. Click **Add connector**, and in the **Connector Plugins** page, click
   **Sample Data**.
6. In the **Launch Sample Data** dialog, click **Users**, and click
   **Launch**.

   It can take a few minutes to create the connector.

<a id="flink-sql-run-snapshot-query-query-topic"></a>

## Step 2: Run a snapshot query on the topic (Flink workspace)

1. In the navigation menu, click **Topics**.
2. In the topics list, find the topic you want to query. If you created a
   Datagen source connector, the topic is named `sample_data_users`.
3. Click the topic name to open the topic details page.
4. Click **Query with Flink**.

   A Flink workspace opens with a SQL editor that you can use to run a snapshot
   query.
5. In the cell, find the **Mode** dropdown, which defaults to **Streaming**.
6. Change the mode to **Snapshot** and click **Run**.

   The query runs and returns all of the messages produced to the topic so
   far.

<a id="flink-sql-run-snapshot-query-set-snapshot-mode"></a>

## Step 3: Set the snapshot mode in SQL (alternative)

When the **Mode** dropdown is not available, you can set the snapshot mode
in SQL by using the
[SET](../reference/statements/set.md#flink-sql-set-statement-config-options) statement to assign the
`sql.snapshot.mode` configuration option.

1. Prepend the SELECT statement with the following SET statement:
   ```sql
   SET 'sql.snapshot.mode' = 'now';
   SELECT * FROM `<your-env>`.`<your-cluster>`.`sample_data_users`;
   ```
2. Click **Run**.

   The query runs and returns all of the messages produced to the topic so
   far.

## Step 4: Invoke snapshot queries programmatically

To enable snapshot queries, set the `sql.snapshot.mode` property to `now`.
You can set this property in the following ways:

- **Table API:** In the `Cloud.Properties` project file, add `sql.snapshot.mode = now`.
- **REST API:** In the statement’s `spec.properties` map, add `"sql.snapshot.mode": "now"`.
- **Terraform:** In the statement properties, add `"sql.snapshot.mode" = "now"`.
- **Confluent CLI:** In the property map, add `--property "sql.snapshot.mode"="now"`

Using curl:

> ```bash
> # Your Confluent Cloud IDs
> ## The endpoint is composed of the target region (here us-east-1.aws)
> export CCLOUD_FLINK_API_ENDPOINT="https://flink.us-east-1.aws.confluent.cloud"
> export CCLOUD_ORGANIZATION_ID=""
> export CCLOUD_ENVIRONMENT_ID="env-..."
> export CCLOUD_FLINK_COMPUTE_POOL_ID="lfcp-..."

> # Your Confluent Cloud API Key and Secret
> export CCLOUD_API_KEY=""
> export CCLOUD_API_SECRET=""
> export CCLOUD_API_BASE64=$(echo -n "${CCLOUD_API_KEY}:${CCLOUD_API_SECRET}" | base64)

> # Your Flink parameters, including the SQL script without any SET command
> export SQL_CATALOG="env-..."
> export SQL_DATABASE="lkc-..."
> export SQL_STATEMENT_NAME="name001"
> export SQL_STATEMENT="SELECT 1;"

> # Call : Note how in the data.spec.properties we can pass SET commands directly, here sql.snapshot.mode=now

> curl --request POST \
> --url "${CCLOUD_FLINK_API_ENDPOINT}/sql/v1/organizations/${CCLOUD_ORGANIZATION_ID}/environments/${CCLOUD_ENVIRONMENT_ID}/statements" \
> --header "Authorization: Basic ${CCLOUD_API_BASE64}" \
> --header "content-type: application/json" \
> --data "{\"name\":\"${SQL_STATEMENT_NAME}\",\"organization_id\":\"${CCLOUD_ORGANIZATION_ID}\",\"environment_id\":\"${CCLOUD_ENVIRONMENT_ID}\",\"spec\":{\"statement\":\"${SQL_STATEMENT}\",\"properties\":{\"sql.snapshot.mode\":\"now\",\"sql.current-catalog\":\"${SQL_CATALOG}\",\"sql.current-database\":\"${SQL_DATABASE}\"},\"compute_pool_id\":\"${CCLOUD_FLINK_COMPUTE_POOL_ID}\"}}"

> # Checking the status of that statement (by name)

> curl -X GET \
> --url "${CCLOUD_FLINK_API_ENDPOINT}/sql/v1/organizations/${CCLOUD_ORGANIZATION_ID}/environments/${CCLOUD_ENVIRONMENT_ID}/statements/${SQL_STATEMENT_NAME}" \
> --header "Authorization: Basic ${CCLOUD_API_BASE64}" \
> --header "content-type: application/json"
> ```

Using the Confluent CLI:

> ```bash
> # Your Flink parameters, including the SQL script without any SET command
> export SQL_CATALOG="env-..."
> export SQL_COMPUTE="lfcp-..."
> export SQL_DATABASE="lkc-..."
> export SQL_STATEMENT_NAME="name001"
> export SQL_STATEMENT="SELECT 1;"

> confluent flink statement create "${SQL_STATEMENT_NAME}" \
> --environment "${SQL_CATALOG}" \
> --compute-pool "${SQL_COMPUTE}" \
> --database "${SQL_DATABASE}" \
> --sql "${SQL_STATEMENT}" \
> --property "sql.snapshot.mode"="now"
> ```

## Related content

- [Snapshot Queries](../concepts/snapshot-queries.md#flink-sql-snapshot-queries)
- [Query Tableflow Tables with Flink](../../topics/tableflow/how-to-guides/query-engines/query-with-flink.md#cloud-tableflow-query-with-flink)
- [Statements](../concepts/statements.md#flink-sql-statements)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).