业务应用和大数据平台的数据流向

概述

业务应用与大数据平台之间的交互是实现数据驱动决策和实时业务处理的关键环节。其交互方式多样，协议选择取决于数据流向、实时性要求及技术架构。一句话总结，数据流向可以是从业务应用写入大数据平台，也可以是大数据平台回写至业务应用的数据库。

一、交互场景与协议分类

1. 数据采集（业务应用 → 大数据平台）

目标：将业务数据（日志、交易记录等）实时或批量传输到大数据平台存储或处理。

常用协议与工具：

HTTP/HTTPS：
- 场景：业务应用通过 REST API 发送数据。
- 工具：
  - Apache Flume：通过 HTTP Source 接收数据，写入 HDFS/Kafka。
  - 自定义 API 服务：业务应用直接调用大数据平台的 API 网关。
- 示例：
```
# 业务应用通过 POST 请求发送 JSON 数据
import requests
data = {"user_id": 123, "action": "purchase"}
response = requests.post("https://bigdata-api.example.com/events", json=data)
```

消息队列协议（TCP/AMQP/MQTT）：

场景：高吞吐、低延迟的实时数据传输。
工具：
- Apache Kafka：业务应用通过 Kafka Producer 发送数据到 Topic。
- RabbitMQ：使用 AMQP 协议传输数据。

示例（Kafka）：

// 业务应用发送数据到 Kafka
Properties props = new Properties();
props.put("bootstrap.servers", "kafka-broker:9092");
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("user_events", "key", "{\"event\": \"login\"}"));

文件传输协议（SFTP/SCP/HDFS API）：
- 场景：批量上传日志文件或数据库导出文件。
- 工具：
  - Apache NiFi：通过 SFTP 拉取文件后写入 HDFS。
  - Hadoop HDFS Client：直接调用 HDFS API 上传文件。
- 示例（HDFS CLI）：
```
# 业务服务器上传日志到 HDFS
hdfs dfs -put /var/log/app.log /data/raw/logs/
```

2. 数据处理与查询（双向交互）

目标：大数据平台处理数据后，业务应用查询结果或订阅实时分析结果。

常用协议与工具：

SQL 协议（JDBC/ODBC）：

场景：业务应用通过 SQL 查询数据仓库。
工具：
- Apache Hive/Trino：提供 JDBC 驱动，支持标准 SQL 查询。
- ClickHouse：高性能 OLAP 数据库，支持 HTTP 和 JDBC。

示例（JDBC 查询 Hive）：

// 业务应用通过 JDBC 连接 Hive
Class.forName("org.apache.hive.jdbc.HiveDriver");
Connection conn = DriverManager.getConnection("jdbc:hive2://hive-server:10000/default");
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("SELECT user_id, COUNT(*) FROM logs GROUP BY user_id");

REST API：
- 场景：查询预计算的结果（如报表、用户画像）。
- 工具：
  - Elasticsearch：通过 REST API 提供全文检索和聚合结果。
  - Superset/Tableau：可视化工具通过 API 拉取数据。
- 示例（查询 Elasticsearch）：
```
# 业务应用查询用户行为统计
curl -XGET "http://es-server:9200/user_actions/_search?q=action:login"
```
流式结果订阅（WebSocket/SSE）：
- 场景：实时监控或告警（如风控系统接收实时异常事件）。
- 工具：
  - Apache Flink：通过 WebSocket 或 Kafka 推送实时处理结果。
  - Redis Pub/Sub：业务应用订阅频道获取实时数据。
- 示例（Flink + WebSocket）：
```
// Flink 将处理结果写入 WebSocket Sink
DataStream<String> alerts = ...; // 实时风控结果
alerts.addSink(new WebSocketSink("ws://business-app:8080/alerts"));
```

3. 反向数据同步（大数据平台 → 业务应用）

目标：将分析结果（推荐模型、用户标签）回写至业务数据库或缓存。

常用协议与工具：

数据库协议（JDBC/ODBC）：
- 场景：将聚合结果写入 MySQL、PostgreSQL 等业务数据库。
- 工具：
  - Apache Spark：使用 JDBC 写入业务库。
  - Airflow：通过 Python 脚本同步数据。
- 示例（Spark 写 MySQL）：
```
# Spark 将结果写入业务库
df.write.format("jdbc") \
  .option("url", "jdbc:mysql://mysql-host:3306/app_db") \
  .option("dbtable", "user_stats") \
  .save()
```
缓存协议（Redis/Memcached）：
- 场景：实时更新缓存中的用户画像或推荐结果。
- 工具：
  - Apache Flink：直接调用 Redis Sink 更新缓存。
- 示例（Flink + Redis）：
```
// Flink 实时更新 Redis 中的用户积分
DataStream<Tuple2<String, Integer>> userScores = ...;
userScores.addSink(new RedisSink<>(config, new RedisMapper(...)));
```

文件导出（SFTP/HTTP）：

场景：生成 CSV/Excel 报表供业务下载。
工具：
- Apache Airflow：定时导出数据到 SFTP 服务器。
- MinIO：通过预签名 URL 提供临时下载链接。

示例（Airflow SFTP 导出）：

# Airflow 任务将 Hive 查询结果导出到 SFTP
sftp_operator = SFTPOperator(
    task_id="export_report",
    ssh_conn_id="sftp_conn",
    local_filepath="/tmp/report.csv",
    remote_filepath="/reports/report_{{ ds }}.csv"
)

二、协议选型关键因素

实时性要求：
- 实时：Kafka、WebSocket、Redis Pub/Sub。
- 准实时/批量：JDBC、SFTP、HDFS。
数据规模：
- 大文件/高吞吐：HDFS、Kafka。
- 小数据/低延迟：HTTP、gRPC。
安全性：
- 敏感数据：HTTPS、SFTP、Kerberos 认证的 HDFS。
- 公开数据：HTTP、普通 JDBC。
技术栈兼容性：
- Java 生态：优先选 Kafka、HDFS、Hive JDBC。
- Python 生态：多用 REST API、PySpark、Airflow。

三、典型架构示例

电商实时推荐系统

数据采集：
- 用户点击流数据通过 Kafka 实时发送至 Flink。
数据处理：
- Flink 实时计算用户兴趣标签，写入 Redis。
结果反馈：
- 业务应用（推荐服务）通过 Redis 读取标签，生成推荐列表。
离线分析：
- 每日通过 Spark 批量计算历史订单数据，结果写入 MySQL 供运营查看。

日志分析平台

日志收集：
- 业务服务器通过 Filebeat 将日志发送至 Kafka。
存储与处理：
- Kafka 数据落地到 HDFS，由 Spark 进行 ETL。
查询展示：
- 处理后的数据导入 Elasticsearch，前端通过 REST API 查询可视化仪表盘。

四、安全与治理

认证与授权：
- Kafka：SASL/SCRAM 或 SSL 客户端认证。
- HDFS：Kerberos 集成 LDAP/AD。
数据加密：
- 传输层：TLS（如 HTTPS、Kafka SSL）。
- 存储层：HDFS Transparent Encryption。
审计与监控：
- 记录 API 调用日志（如 Elasticsearch Audit Log）。
- 使用 Prometheus + Grafana 监控接口性能。