创建Smart Connect任务

• 创建一个立即启动的转储任务。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67

  package com.huaweicloud.sdk.test; import com.huaweicloud.sdk.core.auth.ICredential; import com.huaweicloud.sdk.core.auth.BasicCredentials; import com.huaweicloud.sdk.core.exception.ConnectionException; import com.huaweicloud.sdk.core.exception.RequestTimeoutException; import com.huaweicloud.sdk.core.exception.ServiceResponseException; import com.huaweicloud.sdk.kafka.v2.region.KafkaRegion; import com.huaweicloud.sdk.kafka.v2.*; import com.huaweicloud.sdk.kafka.v2.model.*; public class CreateConnectorTaskSolution { public static void main(String[] args) { // The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security. // In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment String ak = System.getenv("CLOUD_SDK_AK"); String sk = System.getenv("CLOUD_SDK_SK"); String projectId = "{project_id}"; ICredential auth = new BasicCredentials() .withProjectId(projectId) .withAk(ak) .withSk(sk); KafkaClient client = KafkaClient.newBuilder() .withCredential(auth) .withRegion(KafkaRegion.valueOf("<YOUR REGION>")) .build(); CreateConnectorTaskRequest request = new CreateConnectorTaskRequest(); request.withInstanceId("{instance_id}"); CreateSmartConnectTaskReq body = new CreateSmartConnectTaskReq(); SmartConnectTaskReqSinkConfig sinkTaskbody = new SmartConnectTaskReqSinkConfig(); sinkTaskbody.withConsumerStrategy("earliest") .withDestinationFileType("TEXT") .withDeliverTimeInterval(300) .withAccessKey("********") .withSecretKey("********") .withObsBucketName("obs_bucket") .withObsPath("obsTransfer-1810125534") .withPartitionFormat("yyyy/MM/dd/HH/mm") .withRecordDelimiter("\n") .withStoreKeys(false); body.withSinkTask(sinkTaskbody); body.withSinkType(CreateSmartConnectTaskReq.SinkTypeEnum.fromValue("OBS_SINK")); body.withSourceType(CreateSmartConnectTaskReq.SourceTypeEnum.fromValue("NONE")); body.withTopicsRegex("topic-obs*"); body.withStartLater(false); body.withTaskName("smart-connect-1"); request.withBody(body); try { CreateConnectorTaskResponse response = client.createConnectorTask(request); System.out.println(response.toString()); } catch (ConnectionException e) { e.printStackTrace(); } catch (RequestTimeoutException e) { e.printStackTrace(); } catch (ServiceResponseException e) { e.printStackTrace(); System.out.println(e.getHttpStatusCode()); System.out.println(e.getRequestId()); System.out.println(e.getErrorCode()); System.out.println(e.getErrorMsg()); } } }

• 创建一个稍后启动的Kafka数据复制任务。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70

  package com.huaweicloud.sdk.test; import com.huaweicloud.sdk.core.auth.ICredential; import com.huaweicloud.sdk.core.auth.BasicCredentials; import com.huaweicloud.sdk.core.exception.ConnectionException; import com.huaweicloud.sdk.core.exception.RequestTimeoutException; import com.huaweicloud.sdk.core.exception.ServiceResponseException; import com.huaweicloud.sdk.kafka.v2.region.KafkaRegion; import com.huaweicloud.sdk.kafka.v2.*; import com.huaweicloud.sdk.kafka.v2.model.*; public class CreateConnectorTaskSolution { public static void main(String[] args) { // The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security. // In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment String ak = System.getenv("CLOUD_SDK_AK"); String sk = System.getenv("CLOUD_SDK_SK"); String projectId = "{project_id}"; ICredential auth = new BasicCredentials() .withProjectId(projectId) .withAk(ak) .withSk(sk); KafkaClient client = KafkaClient.newBuilder() .withCredential(auth) .withRegion(KafkaRegion.valueOf("<YOUR REGION>")) .build(); CreateConnectorTaskRequest request = new CreateConnectorTaskRequest(); request.withInstanceId("{instance_id}"); CreateSmartConnectTaskReq body = new CreateSmartConnectTaskReq(); SmartConnectTaskReqSourceConfig sourceTaskbody = new SmartConnectTaskReqSourceConfig(); sourceTaskbody.withCurrentClusterName("A") .withClusterName("B") .withUserName("user1") .withPassword("********") .withSaslMechanism("SCRAM-SHA-512") .withInstanceId("b54c9dd8-********-********") .withDirection("two-way") .withSyncConsumerOffsetsEnabled(false) .withReplicationFactor(3) .withTaskNum(2) .withRenameTopicEnabled(false) .withProvenanceHeaderEnabled(true) .withConsumerStrategy("latest") .withCompressionType("snappy") .withTopicsMapping("topic-sc-1:topic-sc-3,topic-sc-2:topic-sc-4"); body.withSourceTask(sourceTaskbody); body.withSourceType(CreateSmartConnectTaskReq.SourceTypeEnum.fromValue("KAFKA_REPLICATOR_SOURCE")); body.withStartLater(true); body.withTaskName("smart-connect-2"); request.withBody(body); try { CreateConnectorTaskResponse response = client.createConnectorTask(request); System.out.println(response.toString()); } catch (ConnectionException e) { e.printStackTrace(); } catch (RequestTimeoutException e) { e.printStackTrace(); } catch (ServiceResponseException e) { e.printStackTrace(); System.out.println(e.getHttpStatusCode()); System.out.println(e.getRequestId()); System.out.println(e.getErrorCode()); System.out.println(e.getErrorMsg()); } } }

• 创建一个立即启动的转储任务。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52

  # coding: utf-8 import os from huaweicloudsdkcore.auth.credentials import BasicCredentials from huaweicloudsdkkafka.v2.region.kafka_region import KafkaRegion from huaweicloudsdkcore.exceptions import exceptions from huaweicloudsdkkafka.v2 import * if __name__ == "__main__": # The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security. # In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment ak = os.environ["CLOUD_SDK_AK"] sk = os.environ["CLOUD_SDK_SK"] projectId = "{project_id}" credentials = BasicCredentials(ak, sk, projectId) client = KafkaClient.new_builder() \ .with_credentials(credentials) \ .with_region(KafkaRegion.value_of("<YOUR REGION>")) \ .build() try: request = CreateConnectorTaskRequest() request.instance_id = "{instance_id}" sinkTaskbody = SmartConnectTaskReqSinkConfig( consumer_strategy="earliest", destination_file_type="TEXT", deliver_time_interval=300, access_key="********", secret_key="********", obs_bucket_name="obs_bucket", obs_path="obsTransfer-1810125534", partition_format="yyyy/MM/dd/HH/mm", record_delimiter="\n", store_keys=False ) request.body = CreateSmartConnectTaskReq( sink_task=sinkTaskbody, sink_type="OBS_SINK", source_type="NONE", topics_regex="topic-obs*", start_later=False, task_name="smart-connect-1" ) response = client.create_connector_task(request) print(response) except exceptions.ClientRequestException as e: print(e.status_code) print(e.request_id) print(e.error_code) print(e.error_msg)

• 创建一个稍后启动的Kafka数据复制任务。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55

  # coding: utf-8 import os from huaweicloudsdkcore.auth.credentials import BasicCredentials from huaweicloudsdkkafka.v2.region.kafka_region import KafkaRegion from huaweicloudsdkcore.exceptions import exceptions from huaweicloudsdkkafka.v2 import * if __name__ == "__main__": # The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security. # In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment ak = os.environ["CLOUD_SDK_AK"] sk = os.environ["CLOUD_SDK_SK"] projectId = "{project_id}" credentials = BasicCredentials(ak, sk, projectId) client = KafkaClient.new_builder() \ .with_credentials(credentials) \ .with_region(KafkaRegion.value_of("<YOUR REGION>")) \ .build() try: request = CreateConnectorTaskRequest() request.instance_id = "{instance_id}" sourceTaskbody = SmartConnectTaskReqSourceConfig( current_cluster_name="A", cluster_name="B", user_name="user1", password="********", sasl_mechanism="SCRAM-SHA-512", instance_id="b54c9dd8-********-********", direction="two-way", sync_consumer_offsets_enabled=False, replication_factor=3, task_num=2, rename_topic_enabled=False, provenance_header_enabled=True, consumer_strategy="latest", compression_type="snappy", topics_mapping="topic-sc-1:topic-sc-3,topic-sc-2:topic-sc-4" ) request.body = CreateSmartConnectTaskReq( source_task=sourceTaskbody, source_type="KAFKA_REPLICATOR_SOURCE", start_later=True, task_name="smart-connect-2" ) response = client.create_connector_task(request) print(response) except exceptions.ClientRequestException as e: print(e.status_code) print(e.request_id) print(e.error_code) print(e.error_msg)

• 创建一个立即启动的转储任务。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73

  package main import ( "fmt" "github.com/huaweicloud/huaweicloud-sdk-go-v3/core/auth/basic" kafka "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2" "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2/model" region "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2/region" ) func main() { // The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security. // In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment ak := os.Getenv("CLOUD_SDK_AK") sk := os.Getenv("CLOUD_SDK_SK") projectId := "{project_id}" auth := basic.NewCredentialsBuilder(). WithAk(ak). WithSk(sk). WithProjectId(projectId). Build() client := kafka.NewKafkaClient( kafka.KafkaClientBuilder(). WithRegion(region.ValueOf("<YOUR REGION>")). WithCredential(auth). Build()) request := &model.CreateConnectorTaskRequest{} request.InstanceId = "{instance_id}" consumerStrategySinkTask:= "earliest" destinationFileTypeSinkTask:= "TEXT" deliverTimeIntervalSinkTask:= int32(300) accessKeySinkTask:= "********" secretKeySinkTask:= "********" obsBucketNameSinkTask:= "obs_bucket" obsPathSinkTask:= "obsTransfer-1810125534" partitionFormatSinkTask:= "yyyy/MM/dd/HH/mm" recordDelimiterSinkTask:= "\n" storeKeysSinkTask:= false sinkTaskbody := &model.SmartConnectTaskReqSinkConfig{ ConsumerStrategy: &consumerStrategySinkTask, DestinationFileType: &destinationFileTypeSinkTask, DeliverTimeInterval: &deliverTimeIntervalSinkTask, AccessKey: &accessKeySinkTask, SecretKey: &secretKeySinkTask, ObsBucketName: &obsBucketNameSinkTask, ObsPath: &obsPathSinkTask, PartitionFormat: &partitionFormatSinkTask, RecordDelimiter: &recordDelimiterSinkTask, StoreKeys: &storeKeysSinkTask, } sinkTypeCreateSmartConnectTaskReq:= model.GetCreateSmartConnectTaskReqSinkTypeEnum().OBS_SINK sourceTypeCreateSmartConnectTaskReq:= model.GetCreateSmartConnectTaskReqSourceTypeEnum().NONE topicsRegexCreateSmartConnectTaskReq:= "topic-obs*" startLaterCreateSmartConnectTaskReq:= false taskNameCreateSmartConnectTaskReq:= "smart-connect-1" request.Body = &model.CreateSmartConnectTaskReq{ SinkTask: sinkTaskbody, SinkType: &sinkTypeCreateSmartConnectTaskReq, SourceType: &sourceTypeCreateSmartConnectTaskReq, TopicsRegex: &topicsRegexCreateSmartConnectTaskReq, StartLater: &startLaterCreateSmartConnectTaskReq, TaskName: &taskNameCreateSmartConnectTaskReq, } response, err := client.CreateConnectorTask(request) if err == nil { fmt.Printf("%+v\n", response) } else { fmt.Println(err) } }

• 创建一个稍后启动的Kafka数据复制任务。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79

  package main import ( "fmt" "github.com/huaweicloud/huaweicloud-sdk-go-v3/core/auth/basic" kafka "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2" "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2/model" region "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2/region" ) func main() { // The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security. // In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment ak := os.Getenv("CLOUD_SDK_AK") sk := os.Getenv("CLOUD_SDK_SK") projectId := "{project_id}" auth := basic.NewCredentialsBuilder(). WithAk(ak). WithSk(sk). WithProjectId(projectId). Build() client := kafka.NewKafkaClient( kafka.KafkaClientBuilder(). WithRegion(region.ValueOf("<YOUR REGION>")). WithCredential(auth). Build()) request := &model.CreateConnectorTaskRequest{} request.InstanceId = "{instance_id}" currentClusterNameSourceTask:= "A" clusterNameSourceTask:= "B" userNameSourceTask:= "user1" passwordSourceTask:= "********" saslMechanismSourceTask:= "SCRAM-SHA-512" instanceIdSourceTask:= "b54c9dd8-********-********" directionSourceTask:= "two-way" syncConsumerOffsetsEnabledSourceTask:= false replicationFactorSourceTask:= int32(3) taskNumSourceTask:= int32(2) renameTopicEnabledSourceTask:= false provenanceHeaderEnabledSourceTask:= true consumerStrategySourceTask:= "latest" compressionTypeSourceTask:= "snappy" topicsMappingSourceTask:= "topic-sc-1:topic-sc-3,topic-sc-2:topic-sc-4" sourceTaskbody := &model.SmartConnectTaskReqSourceConfig{ CurrentClusterName: &currentClusterNameSourceTask, ClusterName: &clusterNameSourceTask, UserName: &userNameSourceTask, Password: &passwordSourceTask, SaslMechanism: &saslMechanismSourceTask, InstanceId: &instanceIdSourceTask, Direction: &directionSourceTask, SyncConsumerOffsetsEnabled: &syncConsumerOffsetsEnabledSourceTask, ReplicationFactor: &replicationFactorSourceTask, TaskNum: &taskNumSourceTask, RenameTopicEnabled: &renameTopicEnabledSourceTask, ProvenanceHeaderEnabled: &provenanceHeaderEnabledSourceTask, ConsumerStrategy: &consumerStrategySourceTask, CompressionType: &compressionTypeSourceTask, TopicsMapping: &topicsMappingSourceTask, } sourceTypeCreateSmartConnectTaskReq:= model.GetCreateSmartConnectTaskReqSourceTypeEnum().KAFKA_REPLICATOR_SOURCE startLaterCreateSmartConnectTaskReq:= true taskNameCreateSmartConnectTaskReq:= "smart-connect-2" request.Body = &model.CreateSmartConnectTaskReq{ SourceTask: sourceTaskbody, SourceType: &sourceTypeCreateSmartConnectTaskReq, StartLater: &startLaterCreateSmartConnectTaskReq, TaskName: &taskNameCreateSmartConnectTaskReq, } response, err := client.CreateConnectorTask(request) if err == nil { fmt.Printf("%+v\n", response) } else { fmt.Println(err) } }

更多编程语言的SDK代码示例，请参见API Explorer的代码示例页签，可生成自动对应的SDK代码示例。