Este tópico descreve como instalar e usar o SDK Python para o Realtime Compute for Apache Flink.
Pré-requisitos
-
Crie um par de AccessKey. Para mais informações, consulte Criar um par de AccessKey.
NotaPara evitar riscos de segurança associados à exposição do par de AccessKey da sua conta Alibaba Cloud, recomendamos o uso do par de AccessKey de um usuário RAM. Crie um usuário RAM, conceda-lhe as permissões necessárias para acessar o Realtime Compute for Apache Flink e utilize o par de AccessKey desse usuário para chamar o SDK. Para mais informações, consulte os seguintes tópicos:
Para criar um usuário RAM e um par de AccessKey, consulte Criar um usuário RAM ou Criar um par de AccessKey.
Para conceder permissões a um usuário RAM, consulte Autorizar um usuário RAM.
Python 3.6 ou posterior instalado.
Sua conta possui as permissões necessárias. Para mais informações, consulte Gerenciar permissões.
Instale o SDK Python do Flink
Instale o SDK Python usando pip.
-
Ao executar operações como desenvolvimento de jobs e O&M, chame a API do Realtime Compute Development Console. Para detalhes sobre instalação e uso, consulte Development Console SDK Center.
pip3 install alibabacloud_ververica20220718==1.2.1 -
Para visualizar informações de workspace, adquirir workspaces ou ajustar recursos, chame a API do Realtime Compute Selling Console. Para mais informações sobre instalação e uso, consulte o Realtime Compute Selling Console SDK Center.
pip3 install alibabacloud_foasconsole20211028==1.0.2
Testar APIs e gerar exemplos de SDK online
O OpenAPI Explorer simplifica o uso de APIs. Com ele, você faz chamadas de API, gera dinamicamente código de exemplo do SDK e pesquisa rapidamente operações de API para agilizar seu processo de desenvolvimento. Visualize e baixe o código de exemplo do SDK nas páginas de referência da API do console de desenvolvimento e do Realtime Compute Selling Console. Para etapas detalhadas, consulte Inícios rápidos.
Na seção SDK Sample, selecione Python e clique em Download Complete Project para obter o projeto de exemplo completo do SDK para a API.
Exemplos de código
Visualize workspaces adquiridos
Este exemplo mostra como consultar os detalhes de workspaces adquiridos em uma região específica. O seguinte parâmetro de solicitação é obrigatório .
Region: O ID da região. Por exemplo, 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:])Listar deployments
Este exemplo demonstra como listar todos os deployments em um namespace. Os seguintes parâmetros de solicitação são obrigatórios .
workspace: O ID do workspace. Obtenha esse ID no campoResourceIdretornado pelo exemplo Visualize workspaces adquiridos. Exemplo: adf9e5147a****.namespace: O nome do namespace. Exemplo: 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:])Iniciar um job
Este exemplo ilustra como iniciar um job a partir de um deployment em um namespace. Os seguintes parâmetros de solicitação são obrigatórios .
workspace: O ID do workspace. Exemplo: adf9e5147a****.namespace: O nome do namespace. Exemplo: script****-default.deploymentId: O ID do deployment. Obtenha esse ID chamando a operação ListDeployments. Exemplo: 3171d4d1-5952-4d02-b978-e762493b****.kind: O tipo de offset inicial. Valores válidos: NONE (início sem estado), LATEST_SAVEPOINT (início a partir do savepoint mais recente), FROM_SAVEPOINT (início a partir de um savepoint especificado) e LATEST_STATE (início a partir do estado mais recente).
# -*- 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:])Listar jobs
Use este exemplo para listar todos os jobs de um deployment específico. Os seguintes parâmetros de solicitação são obrigatórios .
workspace: O ID do workspace. Exemplo: adf9e5147a****.namespace: O nome do namespace. Exemplo: script****-default.deploymentId: O ID do deployment. Obtenha esse ID chamando a operação ListDeployments. Exemplo: 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:])Parar um job
Veja neste exemplo como parar um job. Os seguintes parâmetros de solicitação são obrigatórios .
workspace: O ID do workspace. Exemplo: adf9e5147a****.namespace: O nome do namespace. Exemplo: script****-default.jobId: O ID do job. Obtenha esse ID chamando a operação ListJobs. Exemplo: 3171d4d1-5952-4d02-b978-e762493b****.stopStrategy: A estratégia de parada. Valores válidos: NONE (para imediatamente), STOP_WITH_SAVEPOINT (cria um savepoint antes de parar) e STOP_WITH_DRAIN (para com 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:])Documentação relacionada
Para detalhes sobre o SDK Java, consulte a Referência do SDK Java.