python 将mysql转为csv、json导入到Doris数据库

2024-08-21 19:20

本文主要是介绍python 将mysql转为csv、json导入到Doris数据库,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

上一篇文章是将mysql导出成csv文件,适合csv不是很大的情况,以下对脚本进行了优化,采用分块读取csv,降低了内存的使用率,提高了传输速度。

from pydoris.doris_client import DorisClient
import requests
import pandas as pd
import numpy as npfe_host = ''
username = ''
passwd = ''
fe_http_port = ""
fe_query_port = ""doris_client = DorisClient(fe_host=fe_host,fe_query_port=fe_query_port,fe_http_port=fe_http_port,username=username,password=passwd,db='zst_cep_model')# 由于导入的csv文件过大,都是五六G以上,所以使用chunksize 分块获取数据进行操作。
for key, chunk in enumerate(pd.read_csv('xxxx.csv', chunksize=10000, dtype=str)):# 这一步也是很迷,需要读csv,然后转为字符串类型csv = chunk.to_csv(header=False, index=False, sep=',').encode('utf-8')# table_name 必须的是database.table的形式success = doris_client.write('xxx.xxx', csv)if success:print("数据写入成功!")else:print("数据写入失败。", key)breakdel csv

后面需要进行增量更新,使用csv太麻烦,想着使用json传输可能会好点,并直接从mysql中获取数据直接传输到doris,以下是脚本。

import pandas as pd
from pydoris.doris_client import *from tools import *# 配置
config = {// Mysql的配置'mysql_config': {'host': '','port': ,'user': '','password': '','database': ''},// Doris的配置'doris_db_config': {'fe_host': '','username': '','passwd': '','fe_http_port': "",'fe_query_port': "",'db': ''},# 要传输的mysql的目标表'mysql_table': '',# 要接收的Doris的目标表'doris_table': '',
}# MysqlDataBaseClass 是自己编写的Mysql工具类,返回的是Mysql连接对象
yp_apidb = MysqlDataBaseClass(host=config['mysql_config']['host'], port=config['mysql_config']['port'], user=config['mysql_config']['user'],password=config['mysql_config']['password'], database=config['mysql_config']['database'])doris_client = DorisClient(fe_host=config['doris_db_config']['fe_host'],fe_query_port=config['doris_db_config']['fe_query_port'],fe_http_port=config['doris_db_config']['fe_http_port'],username=config['doris_db_config']['username'],password=config['doris_db_config']['passwd'],db=config['doris_db_config']['db'])def get_data_from_mysql(page=1, end_page=None, total_row_num=None, page_size=10000, limit_date='2024-08-12'):result = {'total_page': 0, 'total': 0, 'now_page': page, 'data': [], 'code': False, 'msg': ''}if total_row_num is None:select_res = yp_apidb.ExecuteSQL_Select(sql=f'''SELECT count(1) as total_num FROM `{config["mysql_table"]}` where collect_batch_date >= '{limit_date}';''')totalRowsNum = int(select_res[0]['total_num'])else:totalRowsNum = total_row_numif (totalRowsNum % page_size) == 0:totalPages = int(totalRowsNum / page_size)else:totalPages = int((totalRowsNum / page_size) + 1)result['total_page'] = totalPagesresult['total'] = totalRowsNumif end_page and page > end_page:result['msg'] = '已经达到设置的最后一页'return resultif page > totalPages:result['msg'] = '已经是最后一页'return resultstart_num = int((page - 1) * page_size)limit = f'{start_num}, {page_size}'sql = f'''SELECT * FROM `{config["mysql_table"]}` where collect_batch_date >= '{limit_date}' limit {limit};'''data_list = yp_apidb.ExecuteSQL_Select(sql)result['data'] = data_listresult['code'] = Trueresult['msg'] = '获取成功'return resultdef insert_to_doris(data_list):if len(data_list) > 0:df = pd.DataFrame(data_list)json_data = df.to_json(orient='records')options = WriteOptions()options.set_json_format()options.set_option("strip_outer_array", "true")success = doris_client.write(f"{config['doris_table']}", json_data, options=options)if success:return Trueelse:print("数据写入失败。")return Falseif __name__ == '__main__':page = 1total_row_num = Nonelimit_date = '2024-08-01'# 循环获取下一页,从而达到自动翻页的功能while True:res = get_data_from_mysql(page=page, total_row_num=total_row_num, limit_date=limit_date)print(res['msg'], res['total_page'], res['total'], res['now_page'])total_row_num = res['total']if res['code']:data_list = res['data']flage = insert_to_doris(data_list)if flage is False:breakpage += 1else:print(res['msg'], page)break

以上脚本仅供学习参考,仅为实现临时功能而编写,还有优化的空间。

这篇关于python 将mysql转为csv、json导入到Doris数据库的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1094054

相关文章

Python将博客内容html导出为Markdown格式

《Python将博客内容html导出为Markdown格式》Python将博客内容html导出为Markdown格式,通过博客url地址抓取文章,分析并提取出文章标题和内容,将内容构建成html,再转... 目录一、为什么要搞?二、准备如何搞?三、说搞咱就搞!抓取文章提取内容构建html转存markdown

Python获取中国节假日数据记录入JSON文件

《Python获取中国节假日数据记录入JSON文件》项目系统内置的日历应用为了提升用户体验,特别设置了在调休日期显示“休”的UI图标功能,那么问题是这些调休数据从哪里来呢?我尝试一种更为智能的方法:P... 目录节假日数据获取存入jsON文件节假日数据读取封装完整代码项目系统内置的日历应用为了提升用户体验,

SpringBoot实现数据库读写分离的3种方法小结

《SpringBoot实现数据库读写分离的3种方法小结》为了提高系统的读写性能和可用性,读写分离是一种经典的数据库架构模式,在SpringBoot应用中,有多种方式可以实现数据库读写分离,本文将介绍三... 目录一、数据库读写分离概述二、方案一:基于AbstractRoutingDataSource实现动态

Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

《PythonFastAPI+Celery+RabbitMQ实现分布式图片水印处理系统》这篇文章主要为大家详细介绍了PythonFastAPI如何结合Celery以及RabbitMQ实现简单的分布式... 实现思路FastAPI 服务器Celery 任务队列RabbitMQ 作为消息代理定时任务处理完整

使用Jackson进行JSON生成与解析的新手指南

《使用Jackson进行JSON生成与解析的新手指南》这篇文章主要为大家详细介绍了如何使用Jackson进行JSON生成与解析处理,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1. 核心依赖2. 基础用法2.1 对象转 jsON(序列化)2.2 JSON 转对象(反序列化)3.

Python Websockets库的使用指南

《PythonWebsockets库的使用指南》pythonwebsockets库是一个用于创建WebSocket服务器和客户端的Python库,它提供了一种简单的方式来实现实时通信,支持异步和同步... 目录一、WebSocket 简介二、python 的 websockets 库安装三、完整代码示例1.

揭秘Python Socket网络编程的7种硬核用法

《揭秘PythonSocket网络编程的7种硬核用法》Socket不仅能做聊天室,还能干一大堆硬核操作,这篇文章就带大家看看Python网络编程的7种超实用玩法,感兴趣的小伙伴可以跟随小编一起... 目录1.端口扫描器:探测开放端口2.简易 HTTP 服务器:10 秒搭个网页3.局域网游戏:多人联机对战4.

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》:本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.

MySQL双主搭建+keepalived高可用的实现

《MySQL双主搭建+keepalived高可用的实现》本文主要介绍了MySQL双主搭建+keepalived高可用的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,... 目录一、测试环境准备二、主从搭建1.创建复制用户2.创建复制关系3.开启复制,确认复制是否成功4.同

Python使用自带的base64库进行base64编码和解码

《Python使用自带的base64库进行base64编码和解码》在Python中,处理数据的编码和解码是数据传输和存储中非常普遍的需求,其中,Base64是一种常用的编码方案,本文我将详细介绍如何使... 目录引言使用python的base64库进行编码和解码编码函数解码函数Base64编码的应用场景注意