Quick start

Prerequisites

Before you start, make sure the following requirements are met:

• Realtime Compute for Apache Flink is activated and a workspace is created. For more information, see Activate Realtime Compute for Apache Flink.

• Realtime Compute engine VVR 11.8 and later comes with Python 3.9, 3.10, and 3.11 pre-installed. Use one of these versions for your local development environment.

Set up the local environment

1. Create a working directory.

   mkdir dataframe-quickstart
   cd dataframe-quickstart

2. (Optional) Configure the local development environment. The Flink Python DataFrame API is available only in VVR 11.8 and later, and is not provided in open-source Flink. Install the VVR version of the PyFlink dependency:

   pip install "ververica-flink==11.8.0"

   Note

   ververica-flink is an API-Only package that provides only type definitions and interface declarations for local development. Jobs must still be submitted on the Realtime Compute for Apache Flink platform.

Write a DataFrame API job

Create a quickstart.py file in the current directory and add the following content.

import pyflink.dataframe as pf
from pyflink.dataframe import col


def main():
    pf.config.set("parallelism.default", "1")

    sales = pf.from_records(
        [
            (1, "book", 42),
            (2, "book", 78),
            (3, "food", 120),
            (4, "game", 35),
            (5, "book", 52),
        ],
        schema=["user_id", "item", "amount"],
    )

    result = (
        sales.with_column("bonus_amount", col("amount") + 10)
        .filter(col("amount") >= 50)
        .rename({"item": "category"})
        .select("user_id", "category", "amount", "bonus_amount")
    )

    result.write_generic(connector="print", options={"logger": True})


if __name__ == "__main__":
    main()

Deploy the job

After the job is developed, deploy it by performing the following steps:

1. Log on to the Realtime Compute console and go to the target workspace.

2. In the left-side navigation pane, choose File Management and upload quickstart.py.

3. On the O&M Center > Job O&M page, click Deploy Job > Python Job and configure the deployment settings.

   • Deployment Mode: Select Streaming or Batch.

   • Deployment Name: Enter a deployment name, such as quickstart.

   • Engine Version: Select vvr-11.8-jdk11-flink-1.20.

   • Python File URI: Select the uploaded quickstart.py.

4. Click Deploy.

Start the job

1. On the O&M Center > Job O&M page, select Streaming Jobs or Batch Jobs from the drop-down list.

2. In the Actions column of the target job, click Start.

3. In the panel that appears on the right side, click Start.

Verify the results

This job writes results to the Task Manager log using the Print connector. After the job finishes, check the log to verify the output:

1. On the O&M Center > Job O&M page, select Streaming Jobs or Batch Jobs from the drop-down list.

2. Click the name of the target job.

3. Click Job Logs.

4. Click the Task Managers tab and select a task manager from the Current TaskManager drop-down list.

The job runs successfully if the log contains the following data:

+I[2, book, 78, 88]
+I[3, food, 120, 130]
+I[5, book, 52, 62]