流计算

from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.stream.v20190819 import stream_client, models

#设置流名称
stream_name = "my_stream"
#设置数据源
source = "kafka"
#设置读取位置
position = "EARLIEST"

#创建流计算客户端
client = stream_client.StreamClient(credential=cred, region="ap-guangzhou")

try:
    #创建流计算作业
    request = models.CreateStreamJobRequest()
    params = {
        "name": "my_job",
        "processor_type": "python",
        "program": "def process(ctx, data):\n    print(data)\n",
        "resources": {
            "source": {
                "need": True,
                "type": source,
                "input": {
                    "position": position,
                    "data_source": [
                        {
                            "topic": stream_name,
                            "partition": 0
                        }
                    ]
                }
            }
        }
    }
    request.from_json_string(json.dumps(params))

    #创建作业并开启
    response = client.CreateStreamJob(request)
    job_id = response.Job.Id
    client.StartStreamJob(request=models.StartStreamJobRequest(JobId=job_id))

    #查看作业运行状态
    query_request = models.DescribeStreamJobRequest()
    query_request.from_json_string(json.dumps({"job_id": job_id}))
    response = client.DescribeStreamJob(query_request)

    print(response.to_json_string())

except TencentCloudSDKException as error:
    print(error)