StarRocks Spark连接器使用指南：从StarRocks读取数据

StarRocks Spark连接器使用指南：从StarRocks读取数据概述StarRocks提供自研的Spark连接器（StarRocks Connector for Apache Spark），使开发者能够通过Spark框架高效地从StarRocks表中读取数据。该连接器支持三种数据读取方式：Spark SQL、Spark DataFrame和Spark RDD，特别适合需要进行复杂数据..

牧韶希

580人浏览 · 2025-06-04 09:06:33

牧韶希 · 2025-06-04 09:06:33 发布

StarRocks Spark连接器使用指南：从StarRocks读取数据

【免费下载链接】starrocks StarRocks是一个开源的分布式数据分析引擎，用于处理大规模数据查询和分析。 - 功能：分布式数据分析；大规模数据查询；数据分析；数据仓库。 - 特点：高性能；可扩展；易于使用；支持多种数据源。项目地址: https://gitcode.com/GitHub_Trending/st/starrocks

概述

StarRocks提供自研的Spark连接器（StarRocks Connector for Apache Spark），使开发者能够通过Spark框架高效地从StarRocks表中读取数据。该连接器支持三种数据读取方式：Spark SQL、Spark DataFrame和Spark RDD，特别适合需要进行复杂数据处理和机器学习分析的场景。

核心优势

高效数据过滤：支持在StarRocks端进行数据预过滤，减少数据传输量
灵活处理方式：提供SQL、DataFrame和RDD三种API，适应不同开发习惯
分布式并行读取：自动将StarRocks表分区映射为Spark RDD分区
类型系统兼容：完善的StarRocks与Spark数据类型映射关系

版本兼容性

连接器版本	Spark版本	StarRocks版本	Java	Scala
1.1.2	3.2-3.5	2.5+	8	2.12
1.1.1	3.2-3.4	2.5+	8	2.12
1.1.0	3.2-3.4	2.5+	8	2.12
1.0.0	3.x	1.18+	8	2.12
1.0.0	2.x	1.18+	8	2.11

注意：从1.1.1版本开始，连接器不再包含MySQL JDBC驱动，需手动添加到Spark类路径

获取连接器

方式一：下载预编译包

可从Maven中央仓库获取对应版本的JAR包，命名格式为： starrocks-spark-connector-${spark_version}_${scala_version}-${connector_version}.jar

方式二：Maven依赖

<dependency>
  <groupId>com.starrocks</groupId>
  <artifactId>starrocks-spark-connector-3.2_2.12</artifactId>
  <version>1.1.0</version>
</dependency>

方式三：源码编译

下载连接器源码
执行编译命令：
```
sh build.sh 3.2  # 以Spark 3.2为例
```
在target目录获取生成的JAR包

关键参数详解

通用参数

参数	默认值	说明
starrocks.fenodes	-	FE节点HTTP地址，格式`<host>:<http_port>`，多个用逗号分隔
starrocks.table.identifier	-	目标表名，格式`<db>.<table>`
starrocks.request.tablet.size	Integer.MAX_VALUE	每个RDD分区包含的Tablet数量
starrocks.batch.size	4096	单次读取最大行数
starrocks.filter.query	-	StarRocks端过滤条件表达式

Spark SQL/DataFrame特有参数

参数	默认值	说明
starrocks.fe.jdbc.url	-	FE的MySQL协议地址，格式`jdbc:mysql://<host>:<query_port>`
starrocks.user	-	数据库用户名
starrocks.password	-	数据库密码

数据类型映射

1.1.0+版本映射关系

StarRocks类型	Spark类型
BOOLEAN	BooleanType
TINYINT	ByteType
INT	IntegerType
BIGINT	LongType
FLOAT	FloatType
DOUBLE	DoubleType
DECIMAL	DecimalType
VARCHAR	StringType
DATE	DateType
DATETIME	TimestampType

使用示例

前置条件

假设StarRocks中已存在test数据库，用户root有访问权限。

Spark SQL方式

// 创建临时视图
spark.sql(
  """
  |CREATE TEMPORARY VIEW starrocks_table
  |USING starrocks
  |OPTIONS(
  |  "starrocks.fe.http.url"="fe_host:8030",
  |  "starrocks.fe.jdbc.url"="jdbc:mysql://fe_host:9030",
  |  "starrocks.table.identifier"="test.table1",
  |  "starrocks.user"="root",
  |  "starrocks.password"=""
  |)
  |""".stripMargin)

// 执行查询
val df = spark.sql("SELECT * FROM starrocks_table WHERE col1 > 100")

DataFrame方式

val df = spark.read.format("starrocks")
  .option("starrocks.fe.http.url", "fe_host:8030")
  .option("starrocks.fe.jdbc.url", "jdbc:mysql://fe_host:9030")
  .option("starrocks.table.identifier", "test.table1")
  .option("starrocks.user", "root")
  .option("starrocks.password", "")
  .load()

df.filter("col1 > 100").show()