流计算的数据源怎么配置？

流计算的数据源配置需要根据具体的流计算平台和数据源类型进行操作。以下是基于知识库资料的详细配置方法，适用于阿里云相关产品（如物联网边缘计算、智能用户增长等）。

1. 计算源配置

计算源是流计算任务中原始数据的来源，通常用于存储和处理大规模数据。以下以MaxCompute为例说明如何配置计算源：

配置步骤：

1. 进入计算源配置页面
   在Quick Audience或其他支持的平台上，选择配置管理 > 数据源 > 数据源 > 计算源配置。

2. 添加MaxCompute计算源

   • 单击配置计算源按钮。
   • 在弹窗中选择MaxCompute作为计算源类型。

3. 填写参数
   根据实际需求配置以下参数：

   • 显示名称：为计算源命名。
   • AccessKey ID 和 AccessKey Secret：提供访问MaxCompute所需的认证信息。
   • 其他必要参数：如项目名称、Endpoint地址等。

4. 测试连通性

   • 单击连接测试按钮，验证计算源是否能够正常连接。
   • 如果测试失败，请检查参数配置并重新测试。

5. 完成配置

   • 测试通过后，单击添加按钮完成计算源的配置。

注意事项：

• 仅支持一个计算源：当前Quick Audience仅支持接入一个MaxCompute类型的计算源。
• 移除限制：若存在基于该计算源的数据导入任务，则无法直接移除计算源。需先删除相关任务后再进行移除。

2. 维表配置

维表是流计算中用于关联静态数据的表格，通常存储在关系型数据库中。以下以MySQL为例说明如何配置维表：

配置语法：

CREATE TABLE rds_dim_table(
    id int,
    len int,
    content VARCHAR,
    PRIMARY KEY (id),
    PERIOD FOR SYSTEM_TIME
) with (
    type='rds',
    url='jdbc:mysql://your_db_host:your_db_port/your_db_name',
    tableName='your_table_name',
    userName='config://local_db_username',
    password='config://local_db_password'
);

参数说明：

• 主键声明：必须指明维表的主键字段。
• PERIOD FOR SYSTEM_TIME：定义维表的变化周期。
• type：指定数据存储类型，例如rds表示MySQL。
• url：数据库连接地址。
• tableName：数据库中的实际表名。
• userName 和 password：数据库用户名和密码。

注意事项：

• 替换所有占位符（如your_db_host、your_db_name等）为实际值。
• 确保数据库中已存在对应的表结构。

3. 源表配置

源表是流计算任务中流式数据的输入来源，用于驱动流数据分析的运行。以下以EdgeBus为例说明如何配置源表：

配置语法：

create table property (
    propertyName varchar,
    propertyValue varchar,
    productKey varchar,
    deviceName varchar,
    gmtCreate varchar,
    ts varchar,
    tstamp as to_timestamp (cast (ts as bigint)),
    WATERMARK wk FOR tstamp as withOffset (tstamp, 2000)
) with (
    type = 'edgebus',
    jsonParser = 'device_property'
);

参数说明：

• 字段定义：定义源表的字段及其数据类型。
• WATERMARK：用于处理乱序数据，定义时间戳偏移量。
• type：指定源表类型，例如edgebus。
• jsonParser：可选参数，支持device_property、device_event、default三种解析器。

注意事项：

• 每个流计算任务至少需要一个源表。
• 根据实际业务需求调整字段定义和解析器类型。

4. 输出配置

流计算任务的结果需要输出到目标数据存储中，支持多种输出类型（如EdgeBusSink、RDS、File等）。以下分别说明各类型配置方法：

EdgeBusSink 示例：

create table edgehub_output (
    productKey varchar,
    deviceName varchar,
    temperature int,
    productTime timestamp,
    inputTime timestamp,
    outputTime timestamp,
    delayTime bigint
) with (
    type = 'edgebus_sink',
    topic = '/sys/streamCompute/highTemperature'
);

RDS 示例：

create table db (
    productKey varchar,
    deviceName varchar,
    ts timestamp,
    temperature int
) with (
    type = 'rds',
    url='jdbc:mysql://your_db_host:your_db_port/your_db_name',
    tableName='your_table_name',
    userName='config://local_db_username',
    password='config://local_db_password'
);

File 示例：

create table print_sink (
    productKey varchar,
    deviceName varchar,
    eventCode varchar,
    ts timestamp,
    temperature int
) with (
    type = 'file',
    tag = 'prop_filter_b',
    filePath = '/linkedge/run/debug/prop_filter_b.txt'
);

注意事项：

• 根据业务需求选择合适的输出类型。
• 确保目标数据存储的连接信息正确无误。

5. 动态数据源配置

在某些场景下，可能需要动态配置数据源（如工业互联网平台中的组件样式属性）。以下为动态变量的使用方法：

动态OT变量：

1. 选择一个动态变量作为数据源。
2. 配置动态路径参数，支持固定值、页面参数和来自变量三种来源。
3. 查看返回数据，并根据需求添加过滤函数。

动态HTTP变量：

1. 选择一个动态HTTP变量作为数据源。
2. 配置路径参数和接口参数。
3. 查看返回数据，并根据需求修改过滤函数。

总结

以上内容涵盖了流计算中常见的数据源配置方法，包括计算源、维表、源表和输出配置。请根据具体业务需求选择合适的配置方式，并确保所有参数正确无误。