华为云实时流计算服务CS对接与使用完全指南

1. 实时流计算服务CS概述

实时流计算服务（Cloud Stream Service，简称CS）是华为云提供的全托管实时流式大数据分析服务。它运行在公有云上，用户无需感知底层计算集群，只需聚焦于Stream SQL业务逻辑，即可即时执行流数据处理作业。CS服务完全兼容Apache Flink（1.5.3版本）和Apache Spark（2.2.1版本）API，既支持通过在线SQL编辑器快速构建流分析应用，也支持基于开源API进行二次开发，构建自定义Jar包并提交运行。

CS服务的核心定位是降低实时流数据处理的门槛。传统流处理框架如Flink、Spark需要开发者掌握复杂的编程模型和集群运维知识，而CS通过全托管模式和Stream SQL能力，让熟悉SQL的数据分析师也能参与实时计算开发。同时，对于需要复杂处理逻辑的场景，CS保留了基于Flink和Spark API的自定义作业能力，兼顾了易用性与灵活性。

在流式计算领域，“实时”意味着计算框架支持按消息事件逐条处理；“流”指数据如水流般持续不断产生；“计算”则涵盖数学运算、数据分析、算法模型执行等操作。CS服务正是将这三者结合，为用户提供从数据接入、实时计算到结果输出的完整流数据处理链路。

2. 产品优势与核心特性

CS服务具备多项产品优势，使其成为华为云上实时流数据分析的首选服务之一。

2.1 简单易用

CS提供在线SQL编辑平台，用户通过编写Stream SQL即可定义数据流入、数据处理和数据流出逻辑，快速实现业务需求。平台内置了丰富的SQL函数，涵盖统计、异常检测、实时聚类、时间序列分析等场景。用户无需关心计算集群的部署与维护，也无需学习复杂的编程技能，大大降低了流数据分析的开发门槛。

2.2 独享集群与共享集群

CS提供共享集群和独享集群两种运行模式。共享集群适合轻量级、非关键的测试场景，多个用户的作业共享计算资源。独享集群则与其他租户和共享集群完全物理隔离，作业运行不受其他租户干扰，并支持配额限制管理。需要特别注意的是，只有独享集群才可以运行自定义Jar作业。因此，如果用户需要使用自定义作业或对性能隔离有严格要求，必须先创建独享集群。

2.3 按需计费

CS按资源使用量计费，定价单位是SPU（Compute Unit，1 SPU=1核CPU+4G内存）。用户选定SPU数量后按时长计费，用多少算多少，精确到秒。这种灵活的计费模式使得CS非常适合业务量波动较大的流处理场景，用户无需为闲置资源付费。

2.4 高吞吐低时延

CS采用Apache Flink的Dataflow模型，基于高性能计算资源提供完全的实时计算能力。从用户自建的Kafka、MRS-Kafka或DMS-Kafka消费数据时，单SPU每秒吞吐可达1千到2万条消息。配合丰富的IoT SQL函数（如区域检测、偏航检测等），CS在物联网实时监控、车联网数据分析等场景中表现优异。

2.5 丰富的内置功能

CS内置了多项高级分析功能：地理位置分析函数支持对地理空间数据进行实时分析，用户通过SQL即可实现偏航检测、电子围栏等地理分析场景；基于Match Recognize的模式匹配检测帮助业务人员使用SQL实现复杂事件规则检测，典型应用包括欺诈检测、车辆异常行为检测、工业设备异常运行状态检测等；流式机器学习方法支持数据统计、异常检测、实时聚类和时间序列分析。此外，CS还支持通过API网关服务自由访问作业数据，并提供了多种图表类型实时展示数据输出。

3. 开通服务前的准备工作

在使用华为云实时流计算服务CS之前，需要完成一系列准备工作。

需要先登录华为云控制台，点击：华为云控制台，还没有账号，点击：注册并关联，已有账号点击：登录后关联

3.1 华为云账号注册与认证

如果还没有华为云账号，首先需要访问华为云官网完成账号注册。注册时需要提供手机号码、邮箱地址等基本信息，并完成实名认证。企业用户建议完成企业实名认证，以便获得更高的资源配额和更完善的企业级服务支持。

3.2 了解CS服务区域

CS服务在不同区域（Region）提供的能力和资源配额可能有所不同。在选择区域时，建议遵循以下原则：优先选择靠近用户或数据源的地理区域以降低网络延迟；确保所选区域支持CS服务及所需的其他云服务（如DIS、OBS、Kafka等）；考虑数据合规性要求，某些行业数据可能需要存储在特定区域。创建CS实例时，需要指定实例所在的区域和VPC网络。

3.3 获取AK/SK与项目ID

如果计划通过API或SDK方式调用CS服务，需要提前获取访问密钥（AK/SK）和项目ID（Project ID）。AK/SK用于API请求的身份认证，可在华为云控制台的“我的凭证”页面生成。项目ID则用于标识用户所属的具体项目，在不同API调用中作为参数传递。

4. CS集群的创建与管理

CS集群是运行流计算作业的计算资源池。根据作业类型和性能需求，用户可以选择创建共享集群或独享集群。

4.1 共享集群

共享集群是CS服务提供的默认集群模式，用户无需手动创建即可使用。在共享集群中，多个用户的作业共享相同的计算资源。共享集群适合以下场景：开发测试阶段的作业验证；对性能隔离要求不高的轻量级流处理任务；Stream SQL作业的快速原型开发。共享集群无需额外配置，创建作业时直接选择共享集群即可运行。

4.2 独享集群的创建

独享集群适用于生产环境和对性能隔离有严格要求的场景。创建独享集群的步骤如下：

进入CS控制台后，在集群管理页面点击“创建独享集群”。填写集群名称、选择区域和可用区、配置VPC网络和子网信息。根据业务需求选择合适的SPU规格，SPU数量决定了集群的计算能力。独享集群创建完成后，需要进行网络配置，确保集群能够与数据源（如DIS、Kafka）和输出目标（如OBS、RDS）网络互通。如果数据源或输出目标位于不同的VPC中，需要建立VPC对等连接。同时，根据实际需要在安全组中设置相应的入站和出站规则。

4.3 集群的网络连通性配置

网络连通性是CS作业成功运行的关键。在配置网络时，需要关注以下几个方面：

对于Kafka数据源，如果Kafka服务端的端口监听在hostname上，则需要将Kafka Broker节点的hostname和IP对应关系添加到CS集群的hosts文件中。对于MRS集群，需要在CS集群hosts文件中添加MRS集群zookeeper地址的IP和域名映射。如果数据源位于线下IDC或其他云平台，需要通过VPC对等连接或云专线实现网络互通。在配置完成后，可以通过CS提供的网络连通性测试工具验证集群与指定地址是否可达。

5. Flink SQL作业开发

Flink SQL是CS服务最核心的作业开发方式。用户通过编写Stream SQL语句定义数据源、数据处理逻辑和数据输出目标，CS服务自动将SQL编译为Flink流计算作业并执行。

5.1 创建Flink SQL作业

进入CS控制台的作业管理页面，点击“创建作业”。选择作业类型为“Flink SQL作业”，填写作业名称和描述信息。CS提供了丰富的作业模板，包括DIS-CS-OBS样例模板等，用户可以直接基于模板快速创建作业，也可以从空白作业开始编写。

5.2 SQL作业的基本结构

一个完整的Flink SQL作业包含三个核心部分：数据源定义（Source）、数据处理逻辑（Transform）和数据输出定义（Sink）。

数据源定义使用CREATE SOURCE STREAM语句，指定数据来源的类型（如DIS、Kafka）、连接参数和数据格式。数据处理逻辑使用标准的SQL语法，包括SELECT、WHERE、GROUP BY、JOIN、窗口函数等。数据输出定义使用CREATE SINK STREAM语句，指定输出目标的类型和连接参数。

5.3 DIS数据源配置示例

数据接入服务（DIS）是CS最常用的数据源之一。用户数据从DIS接入，CS从DIS的通道读取数据作为作业的输入。以下是一个从DIS读取数据的Source定义示例：

CREATE SOURCE STREAM dis_source (
  car_id STRING,
  car_owner STRING,
  speed DOUBLE,
  timestamp BIGINT
) WITH (
  type = "dis",
  region = "cn-north-1",
  channel = "input-dis",
  partition_count = "1",
  encode = "csv",
  field_delimiter = ","
);

在上述示例中，type指定数据源类型为DIS；region为DIS通道所在的区域；channel为DIS通道名称；partition_count为DIS通道的分区数；encode指定数据编码格式为CSV；field_delimiter为字段分隔符。

5.4 Kafka数据源配置示例

Kafka是另一个广泛使用的数据源。CS支持从用户自建的Kafka集群、华为云MRS-Kafka以及分布式消息服务DMS-Kafka消费数据。以下是从Kafka读取数据的Source定义示例：

CREATE SOURCE STREAM kafka_source (
  user_id STRING,
  action STRING,
  amount DOUBLE,
  event_time BIGINT
) WITH (
  type = "kafka",
  kafka_bootstrap_servers = "ip1:port1,ip2:port2",
  kafka_topic = "input-topic",
  encode = "json"
);

其中kafka_bootstrap_servers为Kafka集群的Broker地址列表；kafka_topic为要消费的Topic名称；encode支持json格式。如果Kafka集群开启了SASL_SSL认证，可以通过kafka_properties参数配置原生属性。

5.5 数据处理逻辑示例

定义好数据源后，就可以编写SQL语句进行实时数据分析。以下示例计算每辆车的实时平均速度，并筛选出超速记录：

CREATE TEMPORARY VIEW car_stats AS
SELECT
  car_id,
  car_owner,
  AVG(speed) OVER (PARTITION BY car_id ORDER BY timestamp RANGE BETWEEN INTERVAL '5' MINUTE PRECEDING AND CURRENT ROW) AS avg_speed_5min,
  MAX(speed) OVER (PARTITION BY car_id ORDER BY timestamp RANGE BETWEEN INTERVAL '5' MINUTE PRECEDING AND CURRENT ROW) AS max_speed_5min,
  COUNT(*) OVER (PARTITION BY car_id ORDER BY timestamp RANGE BETWEEN INTERVAL '5' MINUTE PRECEDING AND CURRENT ROW) AS record_count
FROM dis_source;

CREATE TEMPORARY VIEW over_speed AS
SELECT
  car_id,
  car_owner,
  avg_speed_5min,
  max_speed_5min,
  CASE WHEN max_speed_5min > 120 THEN 'OVER_SPEED' ELSE 'NORMAL' END AS speed_status
FROM car_stats
WHERE max_speed_5min > 120;

CS SQL支持丰富的窗口函数和聚合函数，包括滚动窗口（TUMBLE）、滑动窗口（HOP）、会话窗口（SESSION）等，满足各种实时统计需求。

5.6 数据输出配置示例

处理完成的数据需要输出到目标存储或下游系统。CS支持多种输出目标，包括OBS、RDS、DCS、Kafka、云搜索服务CSS、表格存储服务CloudTable等。

输出到OBS示例：

CREATE SINK STREAM obs_sink (
  car_id STRING,
  car_owner STRING,
  avg_speed DOUBLE,
  max_speed DOUBLE,
  speed_status STRING
) WITH (
  type = "obs",
  region = "cn-north-1",
  bucket = "cs-output-bucket",
  path = "/over-speed-logs/",
  encode = "json"
);

输出到RDS（MySQL）示例：

CREATE SINK STREAM rds_sink (
  car_id STRING,
  car_owner STRING,
  avg_speed DOUBLE,
  alert_time TIMESTAMP
) WITH (
  type = "rds",
  region = "cn-north-1",
  db_url = "jdbc:mysql://192.168.0.100:3306/cs_db",
  username = "cs_user",
  password = "your_password",
  table_name = "over_speed_alerts"
);

输出到Kafka示例：

CREATE SINK STREAM kafka_sink (
  car_id STRING,
  car_owner STRING,
  speed_status STRING
) WITH (
  type = "kafka",
  kafka_bootstrap_servers = "ip1:port1,ip2:port2",
  kafka_topic = "output-topic",
  encode = "json"
);

输出到DCS Redis示例：

CREATE SINK STREAM redis_sink (
  car_id STRING,
  car_owner STRING,
  average_speed DOUBLE,
  total_miles DOUBLE
) WITH (
  type = "dcs_redis",
  cluster_address = "192.168.0.34:6379",
  password = "xxxxxxx",
  value_type = "hash",
  key_value = "$(car_id).hash: (name=$(car_owner), avg_speed=$(average_speed))"
);

5.7 完整Flink SQL作业示例

将以上各部分组合，形成一个完整的Flink SQL作业：

-- 1. 定义数据源：从DIS读取车辆速度数据
CREATE SOURCE STREAM dis_source (
  car_id STRING,
  car_owner STRING,
  speed DOUBLE,
  timestamp BIGINT
) WITH (
  type = "dis",
  region = "cn-north-1",
  channel = "input-dis",
  partition_count = "1",
  encode = "csv",
  field_delimiter = ","
);

-- 2. 数据处理：计算5分钟窗口内的平均速度和最大速度
CREATE TEMPORARY VIEW car_stats AS
SELECT
  car_id,
  car_owner,
  AVG(speed) OVER (PARTITION BY car_id ORDER BY timestamp RANGE BETWEEN INTERVAL '5' MINUTE PRECEDING AND CURRENT ROW) AS avg_speed_5min,
  MAX(speed) OVER (PARTITION BY car_id ORDER BY timestamp RANGE BETWEEN INTERVAL '5' MINUTE PRECEDING AND CURRENT ROW) AS max_speed_5min
FROM dis_source;

-- 3. 筛选超速记录
CREATE TEMPORARY VIEW over_speed AS
SELECT
  car_id,
  car_owner,
  avg_speed_5min,
  max_speed_5min
FROM car_stats
WHERE max_speed_5min > 120;

-- 4. 定义数据输出：将超速记录输出到OBS
CREATE SINK STREAM obs_sink (
  car_id STRING,
  car_owner STRING,
  avg_speed DOUBLE,
  max_speed DOUBLE,
  alert_time TIMESTAMP
) WITH (
  type = "obs",
  region = "cn-north-1",
  bucket = "cs-output-bucket",
  path = "/over-speed-logs/",
  encode = "json"
);

-- 5. 执行输出
INSERT INTO obs_sink
SELECT
  car_id,
  car_owner,
  avg_speed_5min,
  max_speed_5min,
  CURRENT_TIMESTAMP AS alert_time
FROM over_speed;

6. 自定义Jar作业开发

对于Flink SQL无法满足的复杂处理逻辑，CS支持用户基于Flink或Spark API开发自定义Jar作业。自定义作业提供了更高的灵活性和更强的表达能力，适用于算法复杂、需要自定义状态管理或集成第三方库的场景。

6.1 自定义作业的前提条件

开发自定义Jar作业需要满足以下前提条件：已创建运行自定义作业所需的CS独享集群；在构建应用Jar包时，将Spark或Flink的相关依赖排除，避免与CS运行环境冲突；已将自定义Jar包上传到OBS桶中；已创建DIS或其他数据源的输入输出通道并完成配置。

6.2 基于Flink API的开发示例

以下是一个基于Flink API的简单自定义作业示例，从Kafka读取数据，进行简单的过滤和转换后输出到OBS：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.obs.OBSFileSystem;

import java.util.Properties;

public class KafkaToOBSJob {
    public static void main(String[] args) throws Exception {
        // 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000);

        // 配置Kafka消费者
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "192.168.0.100:9092,192.168.0.101:9092");
        kafkaProps.setProperty("group.id", "cs-consumer-group");

        // 从Kafka读取数据
        DataStream<String> sourceStream = env
            .addSource(new FlinkKafkaConsumer<>("input-topic", new SimpleStringSchema(), kafkaProps));

        // 数据处理：过滤包含"ERROR"的日志行
        DataStream<String> filteredStream = sourceStream
            .filter(line -> line.contains("ERROR"))
            .map(line -> line + " [PROCESSED_AT: " + System.currentTimeMillis() + "]");

        // 配置OBS输出
        // 注意：在实际生产环境中，需通过flink-obs-fs-hadoop等连接器实现OBS写入
        // 此处为示意代码，实际使用请参考华为云CS官方文档配置OBS Sink

        // 执行作业
        env.execute("Kafka to OBS Filter Job");
    }
}

6.3 基于Spark API的开发示例

CS同样支持基于Spark API的自定义作业，以下是一个简单的Spark Streaming示例：

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.{Seconds, StreamingContext};
import org.apache.spark.streaming.kafka010._;

object SparkStreamingJob {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("CS-Spark-Streaming-Job")
    val ssc = new StreamingContext(conf, Seconds(10))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.0.100:9092",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id" -> "spark-streaming-group",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("input-topic")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )

    // 数据处理
    stream.map(record => record.value())
          .filter(line => line.contains("ERROR"))
          .print()

    ssc.start()
    ssc.awaitTermination()
  }
}

6.4 自定义作业的提交与运行

自定义作业的提交步骤如下：将开发好的应用打包为Jar文件，确保已排除与CS运行环境冲突的依赖。将Jar包上传到OBS桶中。在CS控制台创建自定义作业，选择作业类型为“自定义作业”，指定Jar包的OBS路径。配置作业的输入输出参数，这些参数会传递给Jar包的main方法。选择已创建的独享集群作为运行集群，提交作业并启动运行。

7. 数据源与输出通道的深度配置

CS作业的输入和输出配置是整个流计算链路的关键环节。本节深入介绍几种常用数据源和输出通道的配置细节。

7.1 DIS数据源的完整配置

DIS（数据接入服务）是华为云提供的流数据采集和分发服务。CS与DIS的集成非常紧密，是实时流数据分析的标准数据接入方案。创建DIS通道时，需要指定通道名称（不可重复，由英文字母、数字、中划线和下划线组成）、数据源类型（如CSV、JSON等）和分区数量。

在CS作业中配置DIS数据源时，除了通道名称和区域外，还可以配置以下高级参数：partition_count指定从哪些分区读取数据；offset指定起始消费位置（最新、最早或指定时间戳）；encode和field_delimiter指定数据解析格式。对于JSON格式的数据，可以使用json_config参数指定字段映射关系。

7.2 Kafka数据源的深度配置

Kafka作为数据源时，需要特别注意网络连通性问题。如果Kafka集群与CS集群不在同一VPC，需要建立VPC对等连接。在配置kafka_bootstrap_servers时，应使用Kafka Broker的内网IP或域名，确保CS集群能够访问。

对于开启了SASL_SSL认证的Kafka集群（如华为云DMS Kafka），可以通过kafka_properties参数配置认证信息：

CREATE SOURCE STREAM kafka_secure_source (
  ...
) WITH (
  type = "kafka",
  kafka_bootstrap_servers = "192.168.0.100:9093",
  kafka_topic = "secure-topic",
  encode = "json",
  kafka_properties = "security.protocol=SASL_SSL;sasl.mechanism=PLAIN;sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"your_username\" password=\"your_password\";"
);

7.3 OBS输出通道的配置

OBS（对象存储服务）是CS最常用的数据输出目标之一。CS可以将作业分析结果输出到OBS，适用于大数据分析、原生云应用程序数据、静态网站托管、备份归档等场景。配置OBS输出时，需要指定桶名称（bucket）和存储路径（path）。CS支持按时间或数据量进行文件滚动，避免单个文件过大。可以通过partition配置实现数据的分区存储，例如按日期分区：

CREATE SINK STREAM obs_partition_sink (
  ...
) WITH (
  type = "obs",
  region = "cn-north-1",
  bucket = "cs-output-bucket",
  path = "/logs/",
  encode = "json",
  partition = "dt"
);

7.4 RDS输出通道的配置

RDS（关系型数据库服务）支持将CS处理结果写入MySQL或PostgreSQL数据库。配置RDS输出时，需要提供数据库连接URL、用户名、密码和目标表名。为了保证写入性能，CS采用批量写入机制，可以通过batch_size参数控制每批写入的记录数。需要注意的是，RDS实例与CS集群必须在同一VPC或通过VPC对等连接互通。

8. 作业的提交、运行与监控

完成作业开发和配置后，即可提交并运行作业。CS提供了完善的作业生命周期管理和监控能力。

8.1 作业提交

在CS控制台的作业编辑页面，完成SQL编写和参数配置后，点击“提交”按钮将作业提交到CS服务。提交前，CS会进行语义校验，检查SQL语法是否正确、数据源和输出通道是否存在且可访问。如果校验通过，作业进入“已提交”状态，等待调度运行。

8.2 作业启动与停止

在作业管理页面，可以对已提交的作业执行启动、停止、暂停和恢复等操作。启动作业时，需要选择运行集群（共享集群或独享集群）和并行度。并行数是指同时运行Flink SQL作业的最大任务数，适度增加并行数会提高作业整体算力，但也需考虑线程增多带来的切换开销。最大并行数不能大于计算单元（CU数量减去管理单元）的4倍。

8.3 作业监控

CS提供了丰富的作业监控能力。在控制台的作业管理页面，可以查看作业的实时监控信息，包括吞吐量（每秒输入/输出记录数）、延迟（数据处理延迟时间）、作业状态（运行中、停止、异常等）。通过监控数据，用户可以及时发现作业异常并进行处理。

8.4 通过Java SDK查询作业监控信息

除了控制台，用户还可以通过CS SDK编程方式查询作业监控信息。以下是通过Java SDK查询作业监控信息的示例代码：

import com.huaweicloud.cs.java.v1.client.ApiClient;
import com.huaweicloud.cs.java.v1.client.ApiException;
import com.huaweicloud.cs.java.v1.MetricApi;

public class JobMonitorExample {
    public static void main(String[] args) {
        // 初始化API客户端
        ApiClient apiClient = new ApiClient();
        apiClient.setBasePath("https://cs.cn-north-1.myhuaweicloud.com");
        // 设置认证信息（AK/SK）
        // ...
        
        MetricApi apiInstance = new MetricApi(apiClient);
        String xProjectId = "your_project_id";
        // 查询多个作业的监控信息
        List<Integer> jobIds = Arrays.asList(12345, 12346);
        
        try {
            QueryJobMetricsResponse result = apiInstance.getJobsMetrics(xProjectId, jobIds);
            System.out.println("监控信息查询成功: " + result);
        } catch (ApiException e) {
            System.err.println("查询监控信息失败: " + e.getResponseBody());
            e.printStackTrace();
        }
    }
}

通过SDK可以获取的监控指标包括总输入速率（totalReadRate，单位：条/秒）、总输出速率（totalWriteRate，单位：条/秒）、各算子的处理记录数和脏数据记录数等。

8.5 通过API Gateway访问作业数据

CS支持通过API Gateway服务对外暴露作业数据。用户可以通过API Gateway接口查询作业的APIG网关服务访问地址，然后将作业数据接入自定义的工作流或可视化系统中。以下是通过Java SDK查询APIG地址的示例：

import com.huaweicloud.cs.java.v1.JobApi;

JobApi apiInstance = new JobApi(apiClient);
String xProjectId = "your_project_id";
Long jobId = 12345L;

try {
    JobApiGSinksResponse result = apiInstance.getApiGSinks(xProjectId, jobId);
    System.out.println("APIG地址: " + result);
} catch (ApiException e) {
    System.err.println("查询APIG地址失败: " + e.getResponseBody());
    e.printStackTrace();
}

9. 常见问题与故障排查

在使用CS服务的过程中，可能会遇到一些常见问题。本节总结了几类典型问题及其解决方法。

9.1 网络连通性问题

CS作业提交失败，日志显示连接超时，通常是由于网络不通导致的。检查CS集群与数据源/输出目标是否在同一VPC。如果不在同一VPC，检查VPC对等连接是否已正确建立并生效。检查安全组规则是否允许CS集群访问数据源/输出目标的端口。对于Kafka数据源，检查hosts文件中是否已正确添加Broker节点的IP和域名映射。CS提供了网络连通性测试工具，可以在集群管理页面测试集群与指定地址的连通性。

9.2 权限与认证问题

作业语义检验时提示DIS通道不存在或OBS Bucket没有授权，通常是由于权限配置不足导致的。检查CS服务是否已获得访问DIS通道和OBS桶的授权。在IAM中为CS服务授权相应的资源访问权限。检查AK/SK是否正确，是否具有调用CS API的权限。对于开启了认证的Kafka或DMS服务，检查认证参数是否正确配置。

9.3 作业性能问题

作业处理速度跟不上数据流入速度，出现数据积压时，可以考虑增加作业的并行度。但需注意最大并行数不能大于计算单元（CU数量减去管理单元）的4倍。升级到更高规格的独享集群，增加SPU数量。优化SQL逻辑，减少不必要的状态存储和复杂计算。检查数据倾斜问题，合理设置分区键。

9.4 数据格式问题

作业运行正常但输出结果不符合预期，检查数据源的数据格式是否与CREATE SOURCE STREAM中定义的字段类型和编码格式一致。检查field_delimiter等解析参数是否正确。对于JSON格式的数据，检查JSON结构是否与定义匹配。查看作业日志中的脏数据记录数（corruptedRecords）指标，定位格式异常的数据。

10. 最佳实践与总结

基于对CS服务的全面了解，本节总结一些实用的最佳实践建议。

在集群选择方面，开发测试阶段可以使用共享集群快速验证逻辑，生产环境建议使用独享集群保证性能隔离和稳定性。如果需要运行自定义Jar作业，必须创建独享集群。

在作业开发方面，优先使用Flink SQL方式开发，降低开发门槛和维护成本。对于SQL无法实现的复杂逻辑，再考虑自定义Jar作业。充分利用CS内置的丰富SQL函数，如地理位置分析、模式匹配、机器学习等。合理设计窗口大小和并行度，平衡计算精度和性能。

在网络配置方面，提前规划VPC网络，确保CS集群与所有数据源和输出目标网络互通。对于跨VPC访问，及时建立VPC对等连接。配置安全组时遵循最小权限原则，只开放必要的端口。

在监控运维方面，启用作业监控和告警，及时发现和处理异常。定期查看作业日志，关注脏数据和异常记录。利用CS SDK或API Gateway将监控数据接入企业运维系统。

华为云实时流计算服务CS为实时流数据处理提供了全托管、易用、高性能的解决方案。无论是通过Stream SQL快速实现数据分析，还是通过Flink/Spark API开发复杂自定义作业，CS都能满足不同层次的需求。希望本文能帮助读者快速掌握CS服务的对接与使用方法，在实际项目中高效落地实时流计算业务。

常见问题解答

问1：CS服务支持哪些数据源？
答：CS支持多种数据源，包括数据接入服务DIS、自建Kafka、华为云MRS-Kafka、分布式消息服务DMS-Kafka等。此外还支持从OBS读取数据作为批处理输入。

问2：CS的共享集群和独享集群有什么区别？
答：共享集群中多个用户的作业共享计算资源，适合开发和测试场景。独享集群与其他租户和共享集群完全物理隔离，作业运行不受其他租户干扰，支持配额限制管理。只有独享集群才能运行自定义Jar作业。

问3：如何解决CS作业提交失败并提示连接超时的问题？
答：连接超时通常是由于网络不通导致的。需要检查CS集群与数据源/输出目标是否在同一VPC，如不在需建立VPC对等连接；检查安全组规则是否允许访问；对于Kafka数据源，检查hosts文件中是否已添加Broker节点的IP和域名映射。

问4：CS作业的并行度如何设置？
答：并行度是指同时运行Flink SQL作业的最大任务数。适度增加并行度会提高作业整体算力，但也需考虑线程增多带来的切换开销。最大并行数不能大于计算单元（CU数量减去管理单元）的4倍。

问5：CS服务如何计费？
答：CS按资源使用量计费，定价单位是SPU（1 SPU=1核CPU+4G内存）。用户选定SPU数量后按时长计费，用多少算多少，精确到秒。

问6：CS自定义作业的开发需要什么前提条件？
答：开发自定义Jar作业需要满足以下条件：已创建CS独享集群；构建Jar包时排除了与CS运行环境冲突的依赖；已将Jar包上传到OBS桶；已创建数据源的输入输出通道。