Livy：Spark REST 服务详解

u012024132

812人浏览 · 2025-09-09 14:17:05

u012024132 · 2025-09-09 14:17:05 发布

一、Livy 架构与核心概念

1.1 Livy 整体架构

text

Livy 架构组成:
┌─────────────────────────────────────────────────┐
│                 Livy Server                     │
├─────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌───────────┐ │
│  │  REST API   │  │ Session Mgmt│  │ Job Mgmt  │ │
│  │   Layer     │  │             │  │           │ │
│  └─────────────┘  └─────────────┘  └───────────┘ │
├─────────────────────────────────────────────────┤
│  ┌─────────────────────────────────────────────┐ │
│  │            Spark Integration                │ │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐    │ │
│  │  │ Interactive │  Batch  │  │  Spark   │    │ │
│  │  │ Sessions  │ │ Jobs    │  │ Context  │    │ │
│  │  └──────────┘ └──────────┘ └──────────┘    │ │
│  └─────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────┤
│  ┌─────────────────────────────────────────────┐ │
│  │         Cluster Manager Integration         │ │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐    │ │
│  │  │  YARN    │ │ Standalone│ │ Kubernetes │  │ │
│  │  └──────────┘ └──────────┘ └──────────┘    │ │
│  └─────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────┘

1.2 核心概念解析

1.2.1 Session 类型

java

public enum SessionKind {
    SPARK,          // 交互式Spark会话
    PYSPARK,        // 交互式PySpark会话  
    SPARKR,         // 交互式SparkR会话
    SQL             // 交互式Spark SQL会话
}

1.2.2 Job 执行模式

java

public class JobSubmitRequest {
    private String file;              // 应用jar包或py文件
    private String className;         // 主类名
    private List<String> args;        // 命令行参数
    private Map<String, String> conf; // Spark配置
    private String driverMemory;      // Driver内存
    private String executorMemory;    // Executor内存
    private Integer numExecutors;     // Executor数量
}

二、Livy 安装与配置

2.1 环境准备与安装

2.1.1 系统要求

bash

# 依赖检查
java -version  # Java 8+
spark-submit --version  # Spark 2.0+
curl --version  # HTTP客户端

# 下载Livy
wget https://archive.apache.org/dist/incubator/livy/0.7.1-incubating/livy-0.7.1-incubating-bin.zip
unzip livy-0.7.1-incubating-bin.zip
cd livy-0.7.1-incubating

2.1.2 基础配置

properties

# livy.conf 基础配置
livy.server.host = 0.0.0.0
livy.server.port = 8998
livy.spark.master = yarn
livy.spark.deploy-mode = cluster

# 安全配置
livy.server.auth.type = kerberos
livy.server.access.control.enabled = true

# 会话管理
livy.server.session.timeout = 2h
livy.server.session.max-creation = 10

2.2 高级配置选项

2.2.1 资源管理配置

properties

# 资源队列配置
livy.server.yarn.queue = default
livy.server.yarn.app-lookup-timeout = 2m

# 动态资源分配
livy.server.dynamic-allocation.enabled = true
livy.server.dynamic-allocation.min-executors = 1
livy.server.dynamic-allocation.max-executors = 50
livy.server.dynamic-allocation.executor-idle-timeout = 60s

# 内存配置
livy.server.spark.driver.memory = 2g
livy.server.spark.executor.memory = 4g
livy.server.spark.executor.cores = 2

2.2.2 高可用配置

properties

# ZooKeeper 高可用
livy.server.ha.enabled = true
livy.server.zk.connection.string = zk1:2181,zk2:2181,zk3:2181
livy.server.zk.session.timeout = 180s

# 状态恢复
livy.server.recovery.state-store = filesystem
livy.server.recovery.state-store.url = hdfs:///livy/recovery

三、REST API 详解

3.1 Session 管理 API

3.1.1 创建会话

bash

# 创建Spark会话
curl -X POST -H "Content-Type: application/json" \
  -d '{
    "kind": "spark",
    "name": "my-spark-session",
    "conf": {
      "spark.executor.memory": "4g",
      "spark.executor.cores": "2"
    },
    "executorMemory": "4g",
    "executorCores": 2,
    "numExecutors": 5
  }' \
  http://livy-server:8998/sessions

# 响应示例
{
  "id": 0,
  "state": "starting",
  "kind": "spark",
  "log": [],
  "appId": null,
  "appInfo": {},
  "owner": "user1"
}

3.1.2 会话状态管理

bash

# 获取会话列表
curl http://livy-server:8998/sessions

# 获取特定会话信息
curl http://livy-server:8998/sessions/0

# 获取会话日志
curl http://livy-server:8998/sessions/0/log

# 删除会话
curl -X DELETE http://livy-server:8998/sessions/0

3.2 代码执行 API

3.2.1 执行Spark代码

bash

# 执行Scala代码
curl -X POST -H "Content-Type: application/json" \
  -d '{
    "code": "val rdd = sc.parallelize(1 to 1000)\nrdd.sum()"
  }' \
  http://livy-server:8998/sessions/0/statements

# 执行SQL查询
curl -X POST -H "Content-Type: application/json" \
  -d '{
    "code": "spark.sql(\"SELECT * FROM table WHERE date = '\''2023-01-01'\''\").show()"
  }' \
  http://livy-server:8998/sessions/0/statements

3.2.2 异步执行与状态查询

bash

# 异步执行代码
response=$(curl -s -X POST -H "Content-Type: application/json" \
  -d '{"code": "sc.parallelize(1 to 1000000).count()"}' \
  http://livy-server:8998/sessions/0/statements)

statement_id=$(echo $response | jq -r '.id')

# 查询执行状态
curl http://livy-server:8998/sessions/0/statements/$statement_id

# 获取执行结果
curl http://livy-server:8998/sessions/0/statements/$statement_id/result

3.3 Batch Job 管理 API

3.3.1 提交批处理作业

bash

# 提交Spark作业
curl -X POST -H "Content-Type: application/json" \
  -d '{
    "file": "hdfs:///apps/spark-job.jar",
    "className": "com.example.SparkJob",
    "args": ["input_path", "output_path"],
    "name": "daily-etl-job",
    "conf": {
      "spark.sql.shuffle.partitions": "200",
      "spark.dynamicAllocation.enabled": "true"
    },
    "driverMemory": "4g",
    "executorMemory": "8g",
    "numExecutors": 10
  }' \
  http://livy-server:8998/batches

# 响应示例
{
  "id": 1,
  "state": "starting",
  "appId": null,
  "appInfo": {},
  "log": [],
  "owner": "user1"
}

3.3.2 批处理作业管理

bash

# 获取批处理作业列表
curl http://livy-server:8998/batches

# 获取特定作业信息
curl http://livy-server:8998/batches/1

# 获取作业日志
curl http://livy-server:8998/batches/1/log

# 删除作业
curl -X DELETE http://livy-server:8998/batches/1

四、多租户与安全配置

4.1 用户认证与授权

4.1.1 Kerberos 集成

properties

# Kerberos 配置
livy.server.auth.type = kerberos
livy.server.auth.kerberos.principal = livy/_HOST@EXAMPLE.COM
livy.server.auth.kerberos.keytab = /etc/security/keytabs/livy.service.keytab
livy.server.auth.kerberos.name.rules = DEFAULT

4.1.2 LDAP 集成

properties

# LDAP 认证配置
livy.server.auth.type = ldap
livy.server.auth.ldap.url = ldap://ldap-server:389
livy.server.auth.ldap.baseDN = ou=users,dc=example,dc=com
livy.server.auth.ldap.uidAttribute = uid

4.2 资源隔离与配额管理

4.2.1 多租户资源隔离

properties

# 基于用户的资源队列映射
livy.server.user.queue.mapping = \
  user1:analytics-queue, \
  user2:batch-queue, \
  *:default-queue

# 用户资源限制
livy.server.user.session.max = 5
livy.server.user.job.max = 10
livy.server.user.memory.max = 16g
livy.server.user.cores.max = 20

4.2.2 队列资源限制

properties

# YARN队列资源配置
livy.server.yarn.queue.capacity.analytics = 40
livy.server.yarn.queue.capacity.batch = 30
livy.server.yarn.queue.capacity.interactive = 30

# 队列级别资源限制
livy.server.queue.analytics.max-cores = 100
livy.server.queue.analytics.max-memory = 200g

五、性能优化与监控

5.1 性能调优配置

5.1.1 连接池优化

properties

# HTTP连接池配置
livy.server.http.max-connections = 100
livy.server.http.max-connections-per-route = 20
livy.server.http.connection-timeout = 30s
livy.server.http.socket-timeout = 300s

# 线程池配置
livy.server.thread-pool.min-size = 10
livy.server.thread-pool.max-size = 100
livy.server.thread-pool.keepalive-time = 60s

5.1.2 会话管理优化

properties

# 会话缓存配置
livy.server.session.cache.enabled = true
livy.server.session.cache.size = 1000
livy.server.session.cache.expiry-time = 30m

# 心跳检测
livy.server.session.heartbeat.timeout = 10m
livy.server.session.heartbeat.interval = 1m

5.2 监控与告警

5.2.1 Prometheus 监控配置

yaml

# prometheus.yml 配置
scrape_configs:
  - job_name: 'livy'
    metrics_path: '/metrics'
    static_configs:
      - targets: ['livy-server:8998']
    relabel_configs:
      - source_labels: [__address__]
        target_label: instance

# 关键监控指标
alerting:
  groups:
  - name: livy-alerts
    rules:
    - alert: LivySessionOverload
      expr: livy_sessions_active > 50
      for: 5m
      labels:
        severity: warning
      annotations:
        summary: "Livy sessions overload"

5.2.2 健康检查端点

bash

# 健康检查
curl http://livy-server:8998/healthz

# 指标端点
curl http://livy-server:8998/metrics

# 服务器信息
curl http://livy-server:8998/info

六、客户端集成与示例

6.1 Python 客户端示例

6.1.1 使用 livy-client

python

from livy import LivySession, LivyBatch

# 创建会话
session = LivySession(
    host="livy-server",
    port=8998,
    kind="pyspark",
    conf={
        "spark.executor.memory": "4g",
        "spark.executor.cores": "2"
    }
)

# 执行代码
result = session.run("""
df = spark.read.parquet("hdfs:///data/events")
df.groupBy("date").count().show()
""")

print(result.output)

# 关闭会话
session.close()

6.1.2 提交批处理作业

python

from livy import LivyClient

client = LivyClient("http://livy-server:8998")

# 提交批处理作业
batch = client.submit_batch(
    file="hdfs:///apps/etl-job.py",
    args=["2023-01-01", "hdfs:///output/20230101"],
    name="daily-etl",
    conf={
        "spark.sql.shuffle.partitions": "200",
        "spark.dynamicAllocation.enabled": "true"
    }
)

# 等待作业完成
batch.wait_for_completion()

# 获取作业状态
print(f"Job state: {batch.state}")
print(f"Application ID: {batch.app_id}")

6.2 Java 客户端示例

6.2.1 使用 Livy Java Client

java

import org.apache.livy.LivyClient;
import org.apache.livy.LivyClientBuilder;

// 创建Livy客户端
LivyClient client = new LivyClientBuilder()
    .setURI(new URI("http://livy-server:8998"))
    .build();

// 提交Scala代码
String code = "val rdd = sc.parallelize(1 to 1000000)\n" +
              "rdd.sum()";

Object result = client.submit(new SparkJob(code)).get();
System.out.println("Result: " + result);

// 提交批处理作业
Map<String, String> conf = new HashMap<>();
conf.put("spark.executor.memory", "4g");
conf.put("spark.executor.cores", "2");

client.submit(new BatchJob()
    .setFile("hdfs:///apps/spark-job.jar")
    .setClassName("com.example.SparkJob")
    .setArgs(Arrays.asList("input", "output"))
    .setConf(conf)
).get();

client.stop(true);

七、故障排查与最佳实践

7.1 常见问题排查

7.1.1 日志分析

bash

# 查看Livy服务器日志
tail -f /var/log/livy/livy-server.log

# 查看会话日志
curl http://livy-server:8998/sessions/0/log

# 查看YARN应用日志
yarn logs -applicationId application_123456789_0001

# 调试模式启动
LIVY_DEBUG=true ./bin/livy-server

7.1.2 性能问题排查

bash

# 监控会话状态
curl http://livy-server:8998/sessions | jq '.sessions[] | {id, state, owner}'

# 检查资源使用
yarn application -list | grep livy

# 网络连接测试
telnet livy-server 8998

7.2 最佳实践指南

7.2.1 会话管理最佳实践

python

# 使用会话池避免频繁创建
class SessionPool:
    def __init__(self, max_sessions=10):
        self.sessions = []
        self.max_sessions = max_sessions
    
    def get_session(self):
        if not self.sessions:
            return self.create_session()
        return self.sessions.pop()
    
    def return_session(self, session):
        if len(self.sessions) < self.max_sessions:
            self.sessions.append(session)
        else:
            session.close()

# 使用上下文管理器确保资源释放
@contextmanager
def livy_session():
    session = LivySession(...)
    try:
        yield session
    finally:
        session.close()

7.2.2 错误处理与重试

python

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def submit_to_livy(code, session_id):
    response = requests.post(
        f"http://livy-server:8998/sessions/{session_id}/statements",
        json={"code": code},
        timeout=30
    )
    response.raise_for_status()
    return response.json()

# 使用示例
try:
    result = submit_to_livy("sc.parallelize(1 to 100).count()", 0)
except requests.exceptions.RequestException as e:
    print(f"Failed to submit code: {e}")
    # 实现重试或fallback逻辑

通过以上详细的配置、API说明和最佳实践，可以充分发挥Livy作为Spark REST服务网关的能力，构建高效、稳定的大数据处理平台。

码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

cover

不想先学一堆 React？用 Codex 帮你完成第一支 Remotion 动效视频

码道开发者社区

cover

突发！英伟达限制员工使用Claude；程序员一条代码删掉89TB数据获刑：被判五年十个月；曝iPhone Duo有望支持微信双开 | 极客头条

码道开发者社区

cover

每日热评｜学术牛马的救星还是幻觉放大器？47k星学术研究Agent技能包深度评测

码道开发者社区

所有评论(0)

查看更多评论

u012024132

已为社区贡献2条内容