Clickhouse: A股分时线bar数据python实验-批量导入[上]

from clickhouse_driver import Client
import pandas as pd
import os
from datetime import datetime, date
import time
import mathdef get_all_files_by_root_sub_dirs(directory, file_type):data = list()if os.path.isdir(directory):  # 是目录dir_list = os.walk(directory)  # os.listdir(directory)for (root, sub, files) in dir_list:for file in files:path = os.path.join(root, file)  # root +'\\'+fileif path.endswith(file_type):data.append(path)else:  # 不是目录，直接输出if directory.endswith(file_type):data.append(directory)return data
def get_code_from_csv_file(file):# .csv; .h5 file# D:\join_quant_data\futures\minute\A.XDCE\A1909.XDCE_2019-07-25_2019-08-12.CSV# D:\join_quant_data\futures\minute\A.XDCE.h5s = os.path.basename(file)  # A1909.XDCE_2019-07-25_2019-08-12.CSVsp = s.split('_')[0]if sp.endswith(".csv") or sp.endswith(".CSV"):code = sp[:-4]else:code = spreturn code  # A1909.XDCE# CREATE TABLE stock_tb
# (
#     `code` String,
#     `datetime` DateTime,
#     `open` Float32,
#     `close` Float32,
#     `low` Float32,
#     `high` Float32,
#     `volume` Float64,
#     `money` Float64,
#     `factor` Float32,
#     `high_limit` Float32,
#     `low_limit` Float32,
#     `avg` Float32,
#     `pre_close` Float32,
#     `paused` Float32,
#     `open_interest` Float64
# )
# ENGINE = MergeTree
# ORDER BY datetimedef insert_data():client = Client('localhost')database_name = "my_db"table_name = 'stock_tb'dir_path = "/mnt/d/join_quant_data_stock_product/stock/minute/"files = get_all_files_by_root_sub_dirs(dir_path,".csv") t0 = time.time()file_num = 0 for _file in files:t_file = time.time()print(f"{_file}  => 第{file_num}个文件， 总共：{len(files)}个！")block_insert_data = [] # 每个文件当批量insert的单元df = pd.read_csv(_file)code = get_code_from_csv_file(_file)for row in df.itertuples():_row = list(row)[0:15]_row[0] = code                                            # code_row[1] = datetime.strptime(row[1],'%Y-%m-%d %H:%M:%S')   # datetime: _row[2] = float(row.open)                                 # open: ,_row[3] = float(row.close)                                # close: ,_row[4] = float(row.low)                                  # low:    _row[5] = float(row.high)                                 # high: ,_row[6] = float(row.volume)                               # volume: ,_row[7] = float(row.money)                                # money: ,_row[8] = float(row.factor)                               # factor: ,_row[9] = float(row.high_limit)                           # high_limit:,_row[10] = float(row.low_limit)                           # low_limit:,_row[11] = float(row.avg)                                 # avg:  ,_row[12] = float(row.pre_close)                           # pre_close:,_row[13] = float(row.paused)                              # paused: ,if math.isnan(row.open_interest):_row[14] = 0.0 else:_row[14] = float(row.open_interest)                   #open_interest:block_insert_data.append(_row)# if file_num ==0:#     print(_row)# 逐条也可以insert json# sql = f"INSERT INTO {database_name}.{table_name} FORMAT JSONEachRow {json.dumps(row_data) * 1}"# 批量insert client.execute(f'INSERT INTO {database_name}.{table_name}  VALUES', block_insert_data,types_check=True)table_info = client.execute(f'select count(1) from {database_name}.{table_name}')print(f"clickhouse stock_tb 表信息: {table_info}")print(f"第{file_num}个文件 总共：{len(files)}个 => {_file}读写完成！ cost time:{time.time()-t_file}")file_num = file_num +1print(f"文件总共：{file_num}读写完成！ cost time:{time.time()-t0}")insert_data()

运行情况如下：

期间clickhouse表数据的动态情况：[不是最后完成入库的情况]

SELECTtable AS `表名`,sum(rows) AS `总行数`,formatReadableSize(sum(data_uncompressed_bytes)) AS `原始大小`,formatReadableSize(sum(data_compressed_bytes)) AS `压缩大小`,round((sum(data_compressed_bytes) / sum(data_uncompressed_bytes)) * 100, 0) AS `压缩率`
FROM system.parts
WHERE table IN ('stock_tb')
GROUP BY table

数据量较大，文件有20万个csv，估计有199G左右，数据导入后，后面便于用于测试查询。
另外，可以看出，clickhouse的压缩率很高，这个优点是空间占用较小。

查看一下表的情况：
折腾了一天一夜，20多万个文件终于导入clickhouse完成。

Clickhouse: A股分时线bar数据python实验-批量导入[上]相关推荐

Python中import导入上一级目录模块及循环import问题的解决
Python中import导入上一级目录模块及循环import问题的解决参考文章: (1)Python中import导入上一级目录模块及循环import问题的解决 (2)https://www.cn ...
python处理大量excel数据-python如何批量处理excel数据？
python如何批量处理excel数据? python批量处理excel数据的方法: excel的内容需要有规律,python才能按顺序读取出来. 比如按行按列读取. 一.安装xlrd模块到pyth ...
微信小程序 -- 数据库数据excel文件批量导入
一.excel文件批量导入数据到数据库 1.创建node.js函数upload,点击upload右键外部终端中安装类库 npm install node-xlsx 2.云函数代码 const clou ...
大一大数据Python实验报告汇总
学生实验报告学生姓名学号组员: 实验项目鸢尾花数据集分类 ■必修 □选修 ■设计性实验 □验证性实验 □操作性实验 □综合性实验实验地点实验仪器台号指导教师实验日期及节次一.实 ...
python导入excel数据-Python数据处理之导入导出excel数据
欢迎点击上方"AntDream"关注我 .Python的一大应用就是数据分析了,而数据分析中,经常碰到需要处理Excel数据的情况.这里做一个Python处理Excel数据的总结, ...
SQL Server中采用BULK INSERT实现大数据量文本文件批量导入
SQL语句: BULK INSERT dbo.table FROM 'e:/test.tbl ' WITH ( ...
mysql怎么批量导入excel数据_phpmyadmin怎么批量导入excel数据到mysql
首先我们得到了一个excel表,里面有很多需要我们导入的数据删除第1行"id""XXX"....的字段行,只保留我们需要的数据部分单击"文件&qu ...
python图像批量导入
场景:以前的几千张旧照片,各种扩展名格式的,通过格式工厂统一转成jpg格式,然后通过python检查是否有拍摄日期,如果没有拍摄日期,则将文件的修改时间做为拍摄日期.GPS时间写入.如果文件名不是以I ...
Python中无法导入上一级目录下的包
parentdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0,parentdir) ...

Clickhouse: A股分时线bar数据python实验-批量导入[上]

Clickhouse: A股分时线bar数据python实验-批量导入[上]相关推荐

最新文章

热门文章