Doris中读取数据。Doris表映射为DataFrame或者RDD，推荐使用DataFrame。Doris端完成数据过滤，减少数据传输量。| Connector | Spark | Doris | Java | Scala |
|---|---|---|---|---|
| 1.0.0 | 2.x | master | 8 | 2.11 |
mvn clean package
编译成功后，会在target目录下生成文件doris-spark-1.0.0-SNAPSHOT.jar。将此文件复制到Spark的ClassPath中即可使用Spark-Doris-Connector。例如，Local模式运行的Spark，将此文件放入jars文件夹下。Yarn集群模式运行的Spark，则将此文件放入预部署包中。
CREATE TEMPORARY VIEW spark_doris USING doris OPTIONS( "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME", "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT", "user"="$YOUR_DORIS_USERNAME", "password"="$YOUR_DORIS_PASSWORD" ); SELECT * FROM spark_doris;
val dorisSparkDF = spark.read.format("doris") .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME") .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT") .option("user", "$YOUR_DORIS_USERNAME") .option("password", "$YOUR_DORIS_PASSWORD") .load() dorisSparkDF.show(5)
import org.apache.doris.spark._ val dorisSparkRDD = sc.dorisRDD( tableIdentifier = Some("$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME"), cfg = Some(Map( "doris.fenodes" -> "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT", "doris.request.auth.user" -> "$YOUR_DORIS_USERNAME", "doris.request.auth.password" -> "$YOUR_DORIS_PASSWORD" )) ) dorisSparkRDD.collect()
| Key | Default Value | Comment |
|---|---|---|
| doris.fenodes | -- | Doris Restful接口地址，支持多个地址，使用逗号分隔 |
| doris.table.identifier | -- | DataFame/RDD对应的Doris表名 |
| doris.request.retries | 3 | 向Doris发送请求的重试次数 |
| doris.request.connect.timeout.ms | 30000 | 向Doris发送请求的连接超时时间 |
| doris.request.read.timeout.ms | 30000 | 向Doris发送请求的读取超时时间 |
| doris.request.query.timeout.s | 3600 | 查询doris的超时时间，默认值为1小时，-1表示无超时限制 |
| doris.request.tablet.size | Integer.MAX_VALUE | 一个RDD Partition对应的Doris Tablet个数。 此数值设置越小，则会生成越多的Partition。 从而提升Spark侧的并行度，但同时会对Doris造成更大的压力。 |
| doris.batch.size | 1024 | 一次从BE读取数据的最大行数。 增大此数值可减少Spark与Doris之间建立连接的次数。 从而减轻网络延迟所带来的的额外时间开销。 |
| doris.exec.mem.limit | 2147483648 | 单个查询的内存限制。默认为 2GB，单位为字节 |
| doris.deserialize.arrow.async | false | 是否支持异步转换Arrow格式到spark-doris-connector迭代所需的RowBatch |
| doris.deserialize.queue.size | 64 | 异步转换Arrow格式的内部处理队列，当doris.deserialize.arrow.async为true时生效 |
| Key | Default Value | Comment |
|---|---|---|
| user | -- | 访问Doris的用户名 |
| password | -- | 访问Doris的密码 |
| doris.filter.query.in.max.count | 100 | 谓词下推中，in表达式value列表元素最大数量。 超过此数量，则in表达式条件过滤在Spark侧处理。 |
| Key | Default Value | Comment |
|---|---|---|
| doris.request.auth.user | -- | 访问Doris的用户名 |
| doris.request.auth.password | -- | 访问Doris的密码 |
| doris.read.field | -- | 读取Doris表的列名列表，多列之间使用逗号分隔 |
| doris.filter.query | -- | 过滤读取数据的表达式，此表达式透传给Doris。 Doris使用此表达式完成源端数据过滤。 |
| Doris Type | Spark Type |
|---|---|
| NULL_TYPE | DataTypes.NullType |
| BOOLEAN | DataTypes.BooleanType |
| TINYINT | DataTypes.ByteType |
| SMALLINT | DataTypes.ShortType |
| INT | DataTypes.IntegerType |
| BIGINT | DataTypes.LongType |
| FLOAT | DataTypes.FloatType |
| DOUBLE | DataTypes.DoubleType |
| DATE | DataTypes.StringType1 |
| DATETIME | DataTypes.StringType1 |
| BINARY | DataTypes.BinaryType |
| DECIMAL | DecimalType |
| CHAR | DataTypes.StringType |
| LARGEINT | DataTypes.StringType |
| VARCHAR | DataTypes.StringType |
| DECIMALV2 | DecimalType |
| TIME | DataTypes.DoubleType |
| HLL | Unsupported datatype |
1: Connector中，将DATE和DATETIME映射为String。由于Doris底层存储引擎处理逻辑，直接使用时间类型时，覆盖的时间范围无法满足需求。所以使用String类型直接返回对应的时间可读文本。