小编点评

**Spark读取doris数据** ```python import pyspark.sql # 连接到doris数据库 spark_client = pyspark.sql.SQLContext.builder.getOrCreate() df = spark_client.read.jdbc( url="jdbc:mysql://localhost:2000/doris_db", table="doris_table", column="item_sku_id", numPartitions=50, ) # 打印数据 print(df) # 关闭连接 spark_client.stop() ``` **参数介绍** * `url`：连接到doris数据库的连接字符串。 * `table`：要读取的doris表的名称。 * `column`：要读取的数据列名。 * `numPartitions`：读取的分区数量。 * `lowerBound`和`upperBound`：指定读取的最小和最大值。 * `properties`：其他连接参数。 **注意** * `doris_db` 是一个用于存储doris数据的数据库名称。 * `doris_table` 是doris数据表的名字。 * `item_sku_id` 是doris数据中的要读取的数据列名。 * `spark_client` 是Spark SQL上下文的实例。

正文

1. 工作中，常常需要将线上doris同步至集市。读取doris数据同读取常规mysql基本相同。如果数据行小于千万，比较简单的方式直接单节点连接、读取和存储。Python示例如下：

def get_data(sql,host='',port=2000,user='',password='',db=''):
    # 支持doris
    import pymysql
    connect = pymysql.connect(host=host,port=port,user=user,password=password,db=db,charset='utf8')
    cursor = connect.cursor()
    cursor.execute('SET query_timeout = 216000;') #单位秒
    cursor.execute(sql)
    result = cursor.fetchall()
    for row in result:
        pass # 存储格式可以自行控制 
    cursor.close()
    connect.close()
    return result

2. 如果数据量比较大，超过千万，甚至过亿，单节点读取会遇到超时以及时效过低的问题。可以使用spark.read.jdbc分布式多节点并发读取。spark读取支持两种方式。

主要参数介绍：

read.jdbc(url=url,table=remote_table,column='item_sku_id',numPartitions=50,lowerBound=lowerBound, upperBound=upperBound,properties=prop)

url：格式如'jdbc:mysql://**.jd.com:2000/数据库名?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&failOverReadOnly=false&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai'

table：可以是表名，也可以是查询sql(也即支持条件查询)，如果是sql，格式如"(SELECT count(*) sku FROM rule_price_result where dt='2023-05-10') AS tmp"

numPartitions：控制并发节点个数

lowerBound+upperBound和properties二选一，控制每个节点读取的数据范围。

lowerBound+upperBound方式：指定读取最低和最高值，spark会结合分区个数和最低最高边界机械做分割。

如果数据分布有倾斜，可以通过predicates列表自行控制范围。

作者：京东零售 赵奇猛

来源：京东云开发者社区