Flink Python SDK examples - Realtime Compute for Apache Flink

This topic describes how to install and use the Python SDK for Realtime Compute for Apache Flink.

Prerequisites

Create an AccessKey pair. For more information, see Create an AccessKey pair.
Note
To avoid the security risk of exposing your Alibaba Cloud account's AccessKey pair, we recommend using the AccessKey pair of a RAM user. Create a RAM user, grant the user the required permissions to access Realtime Compute for Apache Flink, and then use the user's AccessKey pair to call the SDK. For more information, see the following topics:
- To create a RAM user and an AccessKey pair, see Create a RAM user or Create an AccessKey pair.
- To grant permissions to a RAM user, see Authorize a RAM user.
You have Python 3.6 or later installed.
Your account has the required permissions. For more information, see Manage permissions.

Install the Flink Python SDK

Install the Python SDK using pip.

When you perform operations such as job development and O&M, you must call the Realtime Compute Development Console API. For details about installation and usage, see Development Console SDK Center.
```
pip3 install alibabacloud_ververica20220718==1.2.1
```
To view workspace information, purchase workspaces, or adjust resources, you need to call the Realtime Compute Selling Console API. For more information about installation and usage, see the Realtime Compute Selling Console SDK Center.
```
pip3 install alibabacloud_foasconsole20211028==1.0.2
```

Test APIs and generate SDK examples online

OpenAPI Explorer simplifies using APIs. You can use it to make API calls, dynamically generate SDK sample code, and quickly search for API operations to streamline your development process. You can view and download the SDK sample code on the API reference pages for the development console and Realtime Compute Selling Console. For detailed steps, see Quick starts.

In the SDK Sample section, select Python and then click Download Complete Project to get the complete SDK sample project for the API.

Code examples

Note

The endpoints for the Realtime Compute Selling Console are listed in Endpoints.
The endpoints for the development console are listed in Endpoints.

View purchased workspaces

This example shows how to query the details of purchased workspaces in a specified region. The following request parameter is required.

Region: The region ID. For example, cn-hangzhou..

# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_foasconsole20211028.client import Client as foasconsole20211028Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_foasconsole20211028 import models as foasconsole_20211028_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
    def __init__(self):
        pass
    @staticmethod
    def create_client() -> foasconsole20211028Client:
        """
        Use an AccessKey pair to initialize the client.
        @return: Client
        @throws Exception
        """
        # Hard-coding your AccessKey pair into your project code can lead to security risks. We recommend using a more secure method, such as STS. The following code is for reference only.
        config = open_api_models.Config(
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set in your runtime environment.
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set in your runtime environment.
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Modify the endpoint based on your actual requirements.
        config.endpoint = f'foasconsole.aliyuncs.com'
        return foasconsole20211028Client(config)
    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        describe_instances_request = foasconsole_20211028_models.DescribeInstancesRequest(
            region='cn-hangzhou'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # Call the API and print the response.
            response=client.describe_instances_with_options(describe_instances_request, runtime)
            print(response)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        describe_instances_request = foasconsole_20211028_models.DescribeInstancesRequest(
            region='cn-hangzhou'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy this code to run, print the API response yourself.
            await client.describe_instances_with_options_async(describe_instances_request, runtime)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
    Sample.main(sys.argv[1:])

List deployments

This example shows how to list all deployments in a namespace. The following request parameters are required.

workspace: The workspace ID. You can obtain this ID from the ResourceId returned by the View purchased workspaces example. Example: adf9e5147a****.
namespace: The name of the namespace. Example: script****-default.

# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_ververica20220718.client import Client as ververica20220718Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_ververica20220718 import models as ververica_20220718_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
    def __init__(self):
        pass
    @staticmethod
    def create_client() -> ververica20220718Client:
        """
        Use an AccessKey pair to initialize the client.
        @return: Client
        @throws Exception
        """
        # Hard-coding your AccessKey pair into your project code can lead to security risks. We recommend using a more secure method, such as STS. The following code is for reference only.
        config = open_api_models.Config(
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set in your runtime environment.
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set in your runtime environment.
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Modify the endpoint based on your actual requirements.
        config.endpoint = f'ververica.cn-hangzhou.aliyuncs.com'
        return ververica20220718Client(config)
    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        list_deployments_headers = ververica_20220718_models.ListDeploymentsHeaders(
            workspace='workspace'
        )
        list_deployments_request = ververica_20220718_models.ListDeploymentsRequest()
        runtime = util_models.RuntimeOptions()
        try:
            # Call the API and print the response.
            request=client.list_deployments_with_options('namespace', list_deployments_request, list_deployments_headers, runtime)
            print(request)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        list_deployments_headers = ververica_20220718_models.ListDeploymentsHeaders(
            workspace='workspace'
        )
        list_deployments_request = ververica_20220718_models.ListDeploymentsRequest()
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy this code to run, print the API response yourself. The `namespace` parameter specifies the name of the namespace.
            await client.list_deployments_with_options_async('namespace', list_deployments_request, list_deployments_headers, runtime)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
    Sample.main(sys.argv[1:])

Start a job

This example shows how to start a job from a deployment in a namespace. The following request parameters are required.

workspace: The workspace ID. Example: adf9e5147a****.
namespace: The name of the namespace. Example: script****-default.
deploymentId: The deployment ID. You can obtain this ID by calling the ListDeployments operation. Example: 3171d4d1-5952-4d02-b978-e762493b****.
kind: The type of the start offset. Valid values: NONE (stateless start), LATEST_SAVEPOINT (start from the latest savepoint), FROM_SAVEPOINT (start from a specified savepoint), and LATEST_STATE (start from the latest state).

# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_ververica20220718.client import Client as ververica20220718Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_ververica20220718 import models as ververica_20220718_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
    def __init__(self):
        pass
    @staticmethod
    def create_client() -> ververica20220718Client:
        """
        Use an AccessKey pair to initialize the client.
        @return: Client
        @throws Exception
        """
        # Hard-coding your AccessKey pair into your project code can lead to security risks. We recommend using a more secure method, such as STS. The following code is for reference only.
        config = open_api_models.Config(
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set in your runtime environment.
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set in your runtime environment.
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Modify the endpoint based on your actual requirements.
        config.endpoint = f'ververica.cn-hangzhou.aliyuncs.com'
        return ververica20220718Client(config)
    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        start_job_with_params_headers = ververica_20220718_models.StartJobWithParamsHeaders(
            workspace='workspace'
        )
        job_start_parameters_deployment_restore_strategy = ververica_20220718_models.DeploymentRestoreStrategy(
            kind='NONE'
        )
        job_start_parameters = ververica_20220718_models.JobStartParameters(
            deployment_id='deploymentId',
            restore_strategy=job_start_parameters_deployment_restore_strategy
        )
        start_job_with_params_request = ververica_20220718_models.StartJobWithParamsRequest(
            body=job_start_parameters
        )
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy this code to run, print the API response yourself.
            client.start_job_with_params_with_options('namespace', start_job_with_params_request, start_job_with_params_headers, runtime)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        start_job_with_params_headers = ververica_20220718_models.StartJobWithParamsHeaders(
            workspace='workspace'
        )
        job_start_parameters_deployment_restore_strategy = ververica_20220718_models.DeploymentRestoreStrategy(
            # The restore strategy for the job.
            kind='NONE'
        )
        job_start_parameters = ververica_20220718_models.JobStartParameters(
            deployment_id='deploymentId',
            restore_strategy=job_start_parameters_deployment_restore_strategy
        )
        start_job_with_params_request = ververica_20220718_models.StartJobWithParamsRequest(
            body=job_start_parameters
        )
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy this code to run, print the API response yourself.
            await client.start_job_with_params_with_options_async('namespace', start_job_with_params_request, start_job_with_params_headers, runtime)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
    Sample.main(sys.argv[1:])

List jobs

This example shows how to list all jobs for a specific deployment. The following request parameters are required.

workspace: The workspace ID. Example: adf9e5147a****.
namespace: The name of the namespace. Example: script****-default.
deploymentId: The deployment ID. You can obtain this ID by calling the ListDeployments operation. Example: 3171d4d1-5952-4d02-b978-e762493b****.

# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_ververica20220718.client import Client as ververica20220718Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_ververica20220718 import models as ververica_20220718_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
    def __init__(self):
        pass
    @staticmethod
    def create_client() -> ververica20220718Client:
        """
        Use an AccessKey pair to initialize the client.
        @return: Client
        @throws Exception
        """
        # Hard-coding your AccessKey pair into your project code can lead to security risks. We recommend using a more secure method, such as STS. The following code is for reference only.
        config = open_api_models.Config(
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set in your runtime environment.
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set in your runtime environment.
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Modify the endpoint based on your actual requirements.
        config.endpoint = f'ververica.cn-hangzhou.aliyuncs.com'
        return ververica20220718Client(config)
    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        list_jobs_headers = ververica_20220718_models.ListJobsHeaders(
            workspace='workspace'
        )
        list_jobs_request = ververica_20220718_models.ListJobsRequest(
            deployment_id='deploymentId'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # Call the API and print the response.
            request=client.list_jobs_with_options('namespace', list_jobs_request, list_jobs_headers, runtime)
            print(request)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        list_jobs_headers = ververica_20220718_models.ListJobsHeaders(
            workspace='workspace'
        )
        list_jobs_request = ververica_20220718_models.ListJobsRequest(
            deployment_id='deploymentId'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy this code to run, print the API response yourself.
            await client.list_jobs_with_options_async('namespace', list_jobs_request, list_jobs_headers, runtime)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
    Sample.main(sys.argv[1:])

Stop a job

This example shows how to stop a job. The following request parameters are required.

workspace: The workspace ID. Example: adf9e5147a****.
namespace: The name of the namespace. Example: script****-default.
jobId: The job ID. You can obtain this ID by calling the ListJobs operation. Example: 3171d4d1-5952-4d02-b978-e762493b****.
stopStrategy: The stop strategy. Valid values: NONE (stops immediately), STOP_WITH_SAVEPOINT (creates a savepoint before stopping), and STOP_WITH_DRAIN (stops with drain).

# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_ververica20220718.client import Client as ververica20220718Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_ververica20220718 import models as ververica_20220718_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
    def __init__(self):
        pass
    @staticmethod
    def create_client() -> ververica20220718Client:
        """
        Use an AccessKey pair to initialize the client.
        @return: Client
        @throws Exception
        """
        # Hard-coding your AccessKey pair into your project code can lead to security risks. We recommend using a more secure method, such as STS. The following code is for reference only.
        config = open_api_models.Config(
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set in your runtime environment.
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Ensure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set in your runtime environment.
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Modify the endpoint based on your actual requirements.
        config.endpoint = f'ververica.cn-hangzhou.aliyuncs.com'
        return ververica20220718Client(config)
    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        stop_job_headers = ververica_20220718_models.StopJobHeaders(
            workspace='workspace'
        )
        stop_job_request_body = ververica_20220718_models.StopJobRequestBody(
            # The stop strategy for the job.
            stop_strategy='stopStrategy'
        )
        stop_job_request = ververica_20220718_models.StopJobRequest(
            body=stop_job_request_body
        )
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy this code to run, print the API response yourself.
            client.stop_job_with_options('namespace', 'jobId', stop_job_request, stop_job_headers, runtime)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        stop_job_headers = ververica_20220718_models.StopJobHeaders(
            workspace='workspace'
        )
        stop_job_request_body = ververica_20220718_models.StopJobRequestBody(
            stop_strategy='stopStrategy'
        )
        stop_job_request = ververica_20220718_models.StopJobRequest(
            body=stop_job_request_body
        )
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy this code to run, print the API response yourself.
            await client.stop_job_with_options_async('namespace', 'jobId', stop_job_request, stop_job_headers, runtime)
        except Exception as error:
            # This is for demonstration only. Implement proper error handling in your production code and do not ignore exceptions.
            # Error message
            print(error.message)
            # Troubleshooting URL
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
    Sample.main(sys.argv[1:])

Realtime Compute for Apache Flink:Python SDK reference