需配置数据库连接与写入逻辑以实现数据持久化与分析，具体路径包括：一、JDBC直连MySQL写入；二、Batch Job分批聚合写入；三、导出CSV供FineBI分析；四、HTTP API暴露聚合结果。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
如果您在使用 MuleRun 执行集成流程时，希望将流转中的数据持久化至数据库并开展后续分析，则需配置数据库连接与写入逻辑。MuleRun 本身不直接提供内置数据分析引擎，但可通过标准 JDBC 连接、SQL 操作及与外部分析工具（如 FineBI、Tableau）协同实现结构化数据采集与分析准备。以下是完成该目标的多种实操路径：
一、通过 JDBC Connector 直连 MySQL 写入分析中间表
该方法适用于需要将 Mule 流中处理后的业务数据（如订单、用户行为日志）实时落库，供下游 BI 工具查询分析。依赖 Mule 4 的 Database Connector，要求 MySQL 驱动已预置且连接参数正确。
1、在 Anypoint Studio 中打开 Mule 项目，拖拽 Database Connector 到流中指定位置。
2、点击 Connector 配置按钮，在 Connection 标签页选择 “Create new configuration”。
3、填写 JDBC URL：jdbc:mysql://localhost:3306/analysis_db?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true，确保端口、数据库名与实际环境一致。
4、输入用户名与密码，并点击 Test Connection 验证连通性。
5、切换至 Operation 标签页，选择 “Insert” 或 “Bulk Insert”，在 SQL Query 字段中编写 INSERT 语句，例如：INSERT INTO user_click_log (user_id, event_time, page_url, duration_ms) VALUES (#[payload.userId], #[payload.timestamp], #[payload.url], #[payload.duration])。
6、保存配置并部署运行，确认数据库表中可查到新插入记录。
二、使用 Batch Job + Database Connector 实现分批聚合写入
当原始数据量大、需先聚合再写入（如按小时统计 PV/UV），Batch Job 可保障内存可控与事务一致性。该方式避免单次大量 INSERT 失败导致全量重试，适合构建轻量级分析宽表。
1、在流中添加 Batch Job 组件，设置 batch size 为 1000。
2、在 Batch Step 中嵌入 Transform Message，使用 DataWeave 聚合 payload，例如按 date_hour 分组并计算 count() 和 distinctBy('user_id')。
3、在 On Complete 阶段添加 Database Connector，Operation 设置为 “Update” 或 “Insert”，SQL 使用参数化语句：INSERT INTO hourly_summary (date_hour, pv, uv) VALUES (#[payload.dateHour], #[payload.pv], #[payload.uv]) ON DUPLICATE KEY UPDATE pv = VALUES(pv), uv = VALUES(uv)。
4、确保目标表已创建主键（如 date_hour），以支持 ON DUPLICATE KEY UPDATE 语法。
5、启动流并发送批量测试消息，检查数据库是否生成聚合结果行。
三、导出至 CSV 文件再由 FineBI 接入分析
适用于无直连数据库权限、或需保留原始明细供多维下钻的场景。MuleRun 可将流中数据序列化为 CSV 并落地至共享目录，FineBI 通过文件数据源自动识别并建模。
1、在流末尾添加 File Connector，Operation 设置为 “Write”。
2、在 Write 配置中，Directory 设置为 /opt/mule/data/export/，File Name 设置为 click_log_#[now().toString('yyyyMMdd_HHmmss')].csv。
3、前置添加 Transform Message，使用 DataWeave 将 payload 转为 CSV 格式，例如：write(payload, 'application/csv', {header: true})。
4、确保 Mule 运行用户对该目录有读写权限，且目录存在。
5、启动流后，检查目标路径是否生成带时间戳的 CSV 文件。
6、登录 FineBI，新建“文件数据源”，选择该 CSV 文件，启用自动解析列类型，完成数据集导入。
四、通过 HTTP Listener + REST API 暴露聚合结果供外部调用
当分析结果需被其他系统（如前端看板、调度平台）程序化获取时，可将 MuleRun 构建为轻量分析服务端，对外暴露 REST 接口返回 JSON 格式聚合数据。
1、在项目中添加 HTTP Listener，Path 设置为 /api/v1/daily-stats，Methods 限定为 GET。
2、添加 Database Connector，Operation 设置为 “Select”，SQL 填写：SELECT DATE(created_at) as stat_date, COUNT(*) as total_orders, SUM(amount) as revenue FROM orders WHERE created_at >= DATE_SUB(NOW(), INTERVAL 7 DAY) GROUP BY DATE(created_at) ORDER BY stat_date DESC。
3、添加 Transform Message，将数据库返回的 Array
4、添加 Set Payload 组件，内容设为 #[output application/json --- {code: 200, data: payload, timestamp: now()}]。
5、部署后使用 curl 或 Postman 访问 http://localhost:8081/api/v1/daily-stats，验证返回 JSON 是否含预期字段与数值。
