1. 需求分析
需求项 内容
──────────────────────────────────────────────────────
数据量 日均数据量、峰值数据量
实时性 数据处理延迟要求
可靠性 数据可靠性要求
安全性 数据安全要求

2. 技术选型
场景 推荐技术
──────────────────────────────────────────────────────
实时数据采集 Kafka + Flume
实时数据处理 Spark Streaming + Flink
离线数据分析 Hive + Spark SQL
数据存储 HDFS + HBase

3. 资源规划
资源 数量 配置
──────────────────────────────────────────────────────
WebSphere节点 4台 16核/64GB
Hadoop节点 6台 32核/128GB
Kafka节点 3台 16核/64GB
Spark节点 4台 32核/128GB

# 集成路线图

阶段 内容 周期
──────────────────────────────────────────────────────
第1阶段 环境搭建 2周
第2阶段 基础集成 2周
第3阶段 应用开发 4周
第4阶段 测试优化 2周

1. 网络设计
┌─────────────────────────────────────────────────────────┐
│ 网络架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ WebSphere区 │ │
│ │ 192.168.1.0/24 │ │
│ └─────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ 大数据区 │ │
│ │ 192.168.2.0/24 │ │
│ │ Hadoop | Kafka | Spark | HBase │ │
│ └─────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘

2. 安全设计
安全措施 说明
──────────────────────────────────────────────────────
网络隔离 WebSphere与大数据网络隔离
访问控制 基于角色的访问控制
数据加密 传输加密、存储加密
认证授权 Kerberos认证

3. 高可用设计
组件 高可用方案
──────────────────────────────────────────────────────
WebSphere 集群部署
Hadoop NameNode HA
Kafka 多副本
Spark Standby Master

1. 认证授权
方式 说明
──────────────────────────────────────────────────────
Kerberos 大数据平台认证
LDAP 用户管理
RBAC 权限控制

2. 数据安全
措施 说明
──────────────────────────────────────────────────────
传输加密 SSL/TLS加密
存储加密 数据加密存储
脱敏处理 敏感数据脱敏

3. 网络安全
措施 说明
──────────────────────────────────────────────────────
防火墙 端口访问控制
VPN 远程访问加密
网络隔离 区域隔离

# 安全配置示例

# Kerberos配置
[libdefaults]
default_realm = FGEDU.NET.CN

[realms]
FGEDU.NET.CN = {
kdc = kdc.fgedu.net.cn
admin_server = kdc.fgedu.net.cn
}

1. 性能指标
指标 目标值
──────────────────────────────────────────────────────
数据吞吐量 100MB/s
查询响应时间 <5秒 数据处理延迟 <10秒 并发连接数 100 2. 性能优化 优化项 措施 ────────────────────────────────────────────────────── 连接池 配置合适的连接池大小 批量处理 使用批量操作 缓存 使用缓存减少查询 压缩 数据压缩传输 3. 资源配置 组件 资源配置 ────────────────────────────────────────────────────── WebSphere JVM堆内存8GB Kafka 堆内存4GB，分区数10 Spark Executor内存16GB HBase RegionServer内存32GB

1. 配置Hive JDBC数据源
# 登录WebSphere管理控制台
https://fgedu-mgr.fgedu.net.cn:9043/ibm/console

# 配置JDBC提供程序
资源 → JDBC → JDBC提供程序 → 新建

名称: Hive JDBC Provider
实现类名: org.apache.hive.jdbc.HiveDriver
类路径: /WebSphere/lib/hive-jdbc-3.1.2.jar

# 配置数据源
资源 → JDBC → 数据源 → 新建

名称: fgedu-hive-ds
JNDI名: jdbc/fgeduhive
数据存储帮助程序类: com.ibm.websphere.rsadapter.ConnectJDBCDataStoreHelper

# 配置连接属性
URL: jdbc:hive2://hadoop.fgedu.net.cn:10000/fgedudb
用户名: fgedu
密码: fgedu123

# 测试连接
点击”测试连接”按钮

执行结果：
连接成功

2. 使用JDBC查询Hive
# Java代码示例
import java.sql.*;

public class HiveQuery {
public static void main(String[] args) {
Connection conn = null;
Statement stmt = null;

try {
// 获取连接
Context ctx = new InitialContext();
DataSource ds = (DataSource) ctx.lookup(“jdbc/fgeduhive”);
conn = ds.getConnection();

// 执行查询
stmt = conn.createStatement();
String sql = “SELECT * FROM fgedu_users LIMIT 10”;
ResultSet rs = stmt.executeQuery(sql);

// 处理结果
while (rs.next()) {
System.out.println(rs.getString(“user_id”) + ” | ” + rs.getString(“user_name”));
}

} catch (Exception e) {
e.printStackTrace();
} finally {
if (stmt != null) stmt.close();
if (conn != null) conn.close();
}
}
}

执行结果：
U001 | 张三
U002 | 李四
U003 | 王五
…

3. HDFS文件操作
# 使用WebHDFS REST API
# 上传文件
curl -i -X PUT “http://hadoop.fgedu.net.cn:9870/webhdfs/v1/data/fgedu-data.csv?op=CREATE”

执行结果：
HTTP/1.1 307 Temporary Redirect
Location: http://hadoop.fgedu.net.cn:9864/webhdfs/v1/data/fgedu-data.csv?op=CREATE…

# 读取文件
curl -i “http://hadoop.fgedu.net.cn:9870/webhdfs/v1/data/fgedu-data.csv?op=OPEN”

执行结果：
HTTP/1.1 200 OK
Content-Type: application/octet-stream

user_id,user_name,amount
U001,张三,1000
U002,李四,2000
…

1. 配置Kafka JMS资源
# 登录WebSphere管理控制台

# 配置JMS提供程序
资源 → JMS → JMS提供程序 → 新建

名称: Kafka JMS Provider
外部初始上下文工厂: io.confluent.kafka.jms.KafkaInitialContextFactory
外部提供程序URL: kafka.fgedu.net.cn:9092

# 配置JMS连接工厂
资源 → JMS → 连接工厂 → 新建

名称: fgedu-kafka-cf
JNDI名: jms/fgedukafka
类型: 队列连接工厂

# 配置JMS目标
资源 → JMS → 目标 → 新建

名称: fgedu-kafka-queue
JNDI名: jms/fgeduqueue
类型: 队列

2. 发送消息到Kafka
# Java代码示例
import javax.jms.*;

public class KafkaProducer {
public void sendMessage(String message) {
Connection connection = null;
Session session = null;
MessageProducer producer = null;

try {
// 获取连接工厂
Context ctx = new InitialContext();
ConnectionFactory cf = (ConnectionFactory) ctx.lookup(“jms/fgedukafka”);
Queue queue = (Queue) ctx.lookup(“jms/fgeduqueue”);

// 创建连接
connection = cf.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
producer = session.createProducer(queue);

// 发送消息
TextMessage textMessage = session.createTextMessage(message);
producer.send(textMessage);

System.out.println(“消息发送成功: ” + message);

} catch (Exception e) {
e.printStackTrace();
} finally {
if (producer != null) producer.close();
if (session != null) session.close();
if (connection != null) connection.close();
}
}
}

执行结果：
消息发送成功: {“userId”:”U001″,”action”:”login”,”timestamp”:”2026-04-10T10:00:00″}

3. 从Kafka接收消息
# Java代码示例
import javax.jms.*;

public class KafkaConsumer implements MessageListener {
public void startConsumer() {
Connection connection = null;
Session session = null;
MessageConsumer consumer = null;

try {
// 获取连接工厂
Context ctx = new InitialContext();
ConnectionFactory cf = (ConnectionFactory) ctx.lookup(“jms/fgedukafka”);
Queue queue = (Queue) ctx.lookup(“jms/fgeduqueue”);

// 创建连接
connection = cf.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
consumer = session.createConsumer(queue);

// 设置监听器
consumer.setMessageListener(this);
connection.start();

System.out.println(“消费者已启动”);

} catch (Exception e) {
e.printStackTrace();
}
}

public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
System.out.println(“收到消息: ” + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}

执行结果：
消费者已启动
收到消息: {“userId”:”U001″,”action”:”login”,”timestamp”:”2026-04-10T10:00:00″}

1. 使用Livy REST API
# Livy是Spark的REST服务

# 提交Spark作业
curl -X POST http://spark.fgedu.net.cn:8998/batches \
-H “Content-Type: application/json” \
-d ‘{
“file”: “/spark/fgedu-analysis.py”,
“className”: “com.fgedu.spark.Analysis”,
“args”: [“–input”, “/data/fgedu-data.csv”, “–output”, “/result/”],
“conf”: {
“spark.executor.memory”: “4g”,
“spark.executor.cores”: “2”
}
}’

执行结果：
{
“id”: 123,
“state”: “starting”,
“appId”: null,
“appInfo”: {},
“log”: []
}

# 查询作业状态
curl http://spark.fgedu.net.cn:8998/batches/123

执行结果：
{
“id”: 123,
“state”: “running”,
“appId”: “application_1612345678_0001”,
“appInfo”: {
“driverLogUrl”: “http://…”,
“sparkUiUrl”: “http://…”
},
“log”: [“Starting Spark job…”]
}

# 获取作业结果
curl http://spark.fgedu.net.cn:8998/batches/123/state

执行结果：
{
“id”: 123,
“state”: “success”
}

2. Java调用Spark REST API
# Java代码示例
import java.net.http.*;
import org.json.*;

public class SparkJobClient {
private String livyUrl = “http://spark.fgedu.net.cn:8998”;

public int submitJob(String jobFile, String[] args) {
HttpClient client = HttpClient.newHttpClient();

// 构建请求体
JSONObject requestBody = new JSONObject();
requestBody.put(“file”, jobFile);
requestBody.put(“args”, new JSONArray(args));

JSONObject conf = new JSONObject();
conf.put(“spark.executor.memory”, “4g”);
requestBody.put(“conf”, conf);

// 发送请求
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(livyUrl + “/batches”))
.header(“Content-Type”, “application/json”)
.POST(HttpRequest.BodyPublishers.ofString(requestBody.toString()))
.build();

HttpResponse response = client.send(request,
HttpResponse.BodyHandlers.ofString());

JSONObject result = new JSONObject(response.body());
return result.getInt(“id”);
}

public String getJobStatus(int jobId) {
HttpClient client = HttpClient.newHttpClient();

HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(livyUrl + “/batches/” + jobId + “/state”))
.build();

HttpResponse response = client.send(request,
HttpResponse.BodyHandlers.ofString());

JSONObject result = new JSONObject(response.body());
return result.getString(“state”);
}
}

执行结果：
作业ID: 123
作业状态: success

1. 使用Phoenix JDBC连接HBase
# 配置Phoenix JDBC数据源

# JDBC提供程序配置
名称: Phoenix JDBC Provider
实现类名: org.apache.phoenix.jdbc.PhoenixDriver
类路径: /WebSphere/lib/phoenix-5.0.0.jar

# 数据源配置
名称: fgedu-hbase-ds
JNDI名: jdbc/fgeduhbase
URL: jdbc:phoenix:hbase.fgedu.net.cn:2181

2. 使用JDBC操作HBase
# Java代码示例
import java.sql.*;

public class HBaseClient {
public void queryData() {
Connection conn = null;
Statement stmt = null;

try {
// 获取连接
Context ctx = new InitialContext();
DataSource ds = (DataSource) ctx.lookup(“jdbc/fgeduhbase”);
conn = ds.getConnection();

// 创建表
stmt = conn.createStatement();
String createSql = “CREATE TABLE IF NOT EXISTS fgedu_users (” +
“user_id VARCHAR PRIMARY KEY,” +
“info.user_name VARCHAR,” +
“info.email VARCHAR” +
“)”;
stmt.execute(createSql);
System.out.println(“表创建成功”);

// 插入数据
String insertSql = “UPSERT INTO fgedu_users VALUES (‘U001’, ‘张三’, ‘zhangsan@fgedu.net.cn’)”;
stmt.execute(insertSql);
conn.commit();
System.out.println(“数据插入成功”);

// 查询数据
String querySql = “SELECT * FROM fgedu_users”;
ResultSet rs = stmt.executeQuery(querySql);

while (rs.next()) {
System.out.println(rs.getString(“user_id”) + ” | ” +
rs.getString(“user_name”) + ” | ” + rs.getString(“email”));
}

} catch (Exception e) {
e.printStackTrace();
} finally {
if (stmt != null) stmt.close();
if (conn != null) conn.close();
}
}
}

执行结果：
表创建成功
数据插入成功
U001 | 张三 | zhangsan@fgedu.net.cn

3. 使用HBase REST API
# 获取数据
curl http://hbase.fgedu.net.cn:8080/fgedu_users/U001

执行结果：



张三

# 扫描表
curl http://hbase.fgedu.net.cn:8080/fgedu_users/*

执行结果：



张三


李四