Bu konuda, Apache Flink için Realtime Compute Python SDK'sının nasıl yükleneceği ve kullanılacağı açıklanmaktadır.
Önkoşullar
-
Bir AccessKey çifti oluşturun. Daha fazla bilgi için bkz. AccessKey çifti oluşturma.
NotAlibaba Cloud hesabınızın AccessKey çiftinin açığa çıkmasından kaynaklanan güvenlik risklerini önlemek için bir RAM kullanıcısının AccessKey çiftini kullanmanızı öneririz. Bir RAM kullanıcısı oluşturun, kullanıcıya Apache Flink için Realtime Compute hizmetine erişmek için gerekli izinleri verin ve ardından SDK'yı çağırmak için bu kullanıcının AccessKey çiftini kullanın. Daha fazla bilgi için aşağıdaki konulara bakın:
Bir RAM kullanıcısı ve AccessKey çifti oluşturmak için bkz. RAM kullanıcısı oluşturma veya AccessKey çifti oluşturma.
Bir RAM kullanıcısına izin vermek için bkz. RAM kullanıcısını yetkilendirme.
Python 3.6 veya sonraki bir sürümü yüklemiş olmanız gerekir.
Hesabınızın gerekli izinlere sahip olması gerekir. Daha fazla bilgi için bkz. İzinleri yönetme.
Flink Python SDK'yı yükleme
Python SDK'yı pip kullanarak yükleyin.
-
İş geliştirme ve işletme ile bakım gibi işlemleri gerçekleştirirken Realtime Compute Development Console API'sini çağırmanız gerekir. Kurulum ve kullanım ayrıntıları için bkz. Geliştirme Konsolu SDK Merkezi.
pip3 install alibabacloud_ververica20220718==1.2.1 -
Çalışma alanı bilgilerini görüntülemek, çalışma alanları satın almak veya kaynakları ayarlamak için Realtime Compute Selling Console API'sini çağırmanız gerekir. Kurulum ve kullanım hakkında daha fazla bilgi için bkz. Realtime Compute Selling Console SDK Merkezi.
pip3 install alibabacloud_foasconsole20211028==1.0.2
API'leri test etme ve çevrimiçi SDK örnekleri oluşturma
OpenAPI Explorer, API kullanımını basitleştirir. API çağrıları yapmak, dinamik olarak SDK örnek kodu oluşturmak ve API işlemlerini hızlıca aramak için bu aracı kullanarak geliştirme sürecinizi hızlandırabilirsiniz. geliştirme konsolu ve Realtime Compute Selling Console API referans sayfalarında SDK örnek kodlarını görüntüleyebilir ve indirebilirsiniz. Ayrıntılı adımlar için bkz. Hızlı başlangıçlar.
SDK Örneği bölümünde Python seçeneğini belirleyin ve ardından API için tam SDK örnek projesini edinmek üzere Tüm Projeyi İndir düğmesine tıklayın.
Kod örnekleri
Realtime Compute Selling Console uç noktaları şu bağlantıda listelenmiştir: Uç Noktalar.
Geliştirme konsolu uç noktaları şu bağlantıda listelenmiştir: Uç Noktalar.
Satın alınan çalışma alanlarını görüntüleme
Bu örnekte, belirtilen bir bölgede satın alınan çalışma alanlarının ayrıntılarının nasıl sorgulanacağı gösterilmektedir. Aşağıdaki istek parametresi gereklidir .
Region: Bölge kimliği. Örneğin, 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:])Dağıtımları listeleme
Bu örnekte, bir ad alanındaki tüm dağıtımların nasıl listeleneceği gösterilmektedir. Aşağıdaki istek parametreleri gereklidir .
workspace: Çalışma alanı kimliği. Bu kimliği, Satın alınan çalışma alanlarını görüntüleme örneğinde döndürülenResourceIddeğerinden elde edebilirsiniz. Örnek: adf9e5147a****.namespace: Ad alanının adı. Örnek: 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:])İş başlatma
Bu örnekte, bir ad alanında bulunan bir dağıtımdan işin nasıl başlatılacağı gösterilmektedir. Aşağıdaki istek parametreleri gereklidir .
workspace: Çalışma alanı kimliği. Örnek: adf9e5147a****.namespace: Ad alanının adı. Örnek: script****-default.deploymentId: Dağıtım kimliği. Bu kimliği ListDeployments işlemi çağırarak elde edebilirsiniz. Örnek: 3171d4d1-5952-4d02-b978-e762493b****.kind: Başlangıç ofsetinin türü. Geçerli değerler: NONE (durumsuz başlangıç), LATEST_SAVEPOINT (en son kaydetme noktasından başlatma), FROM_SAVEPOINT (belirtilen bir kaydetme noktasından başlatma) ve LATEST_STATE (en son durumdan başlatma).
# -*- 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:])İşleri listeleme
Bu örnekte, belirli bir dağıtım için tüm işlerin nasıl listeleneceği gösterilmektedir. Aşağıdaki istek parametreleri gereklidir .
workspace: Çalışma alanı kimliği. Örnek: adf9e5147a****.namespace: Ad alanının adı. Örnek: script****-default.deploymentId: Dağıtım kimliği. Bu kimliği ListDeployments işlemi çağırarak elde edebilirsiniz. Örnek: 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:])İşi durdurma
Bu örnekte, bir işin nasıl durdurulacağı gösterilmektedir. Aşağıdaki istek parametreleri gereklidir .
workspace: Çalışma alanı kimliği. Örnek: adf9e5147a****.namespace: Ad alanının adı. Örnek: script****-default.jobId: İş kimliği. Bu kimliği ListJobs işlemi çağırarak elde edebilirsiniz. Örnek: 3171d4d1-5952-4d02-b978-e762493b****.stopStrategy: Durdurma stratejisi. Geçerli değerler: NONE (hemen durdurur), STOP_WITH_SAVEPOINT (durdurmadan önce bir kaydetme noktası oluşturur) ve STOP_WITH_DRAIN (drain ile durdurur).
# -*- 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:])İlgili belgeler
Java SDK hakkında ayrıntılı bilgi için bkz. Java SDK referansı.