能否提供一个Python脚本的例子,用于从ClickHouse获取数据并将其转换为单行JSON字符串后写入指定的文本文件(.txt格式)到服务器上的特定路径?其中每1w行生成一个文件
时间: 2024-12-26 18:30:18 浏览: 1
当然可以,以下是一个简单的Python脚本示例,使用`clickhouse-driver`库连接ClickHouse数据库,并读取数据生成单行JSON。为了将数据分隔成每1万个行一个文件,我们使用`pandas`库处理数据并按行计数。
```python
import clickhouse_driver
import pandas as pd
from pathlib import Path
import json
# 定义ClickHouse连接信息
CH_HOST = 'your_clickhouse_host'
CH_PORT = your_clickhouse_port
CH_DATABASE = 'your_database_name'
CH_TABLE = 'your_table_name'
# 连接到ClickHouse数据库
client = clickhouse_driver.Client(host=CH_HOST, port=CH_PORT)
def fetch_data():
query = f"SELECT * FROM {CH_DATABASE}.{CH_TABLE}"
# 获取所有数据,注意这可能会非常大,需要考虑内存限制
data = client.execute(query)
# 将结果转化为DataFrame
df = pd.DataFrame(data)
return df
def write_json_to_file(df, file_path):
lines_per_file = 10000
for i in range(0, len(df), lines_per_file):
chunk = df.iloc[i:i+lines_per_file]
# 将DataFrame每一行转换为字典,然后序列化为JSON
json_lines = [json.dumps(row.to_dict()) for _, row in chunk.iterrows()]
with open(file_path.format(i // lines_per_file + 1), 'a') as f:
f.write('\n'.join(json_lines) + '\n')
# 指定输出目录和文件名前缀
output_dir = '/path/to/output/directory/'
file_prefix = 'data_'
# 创建输出目录(如果不存在)
Path(output_dir).mkdir(parents=True, exist_ok=True)
# 开始处理数据并生成文件
for i, df in enumerate(fetch_data().chunksize(lines_per_file)):
output_file = output_dir + file_prefix + str(i+1) + '.txt'
write_json_to_file(df, output_file)
print(f"Wrote {i+1}th file to {output_file}")
```
别忘了替换上述代码中的`your_clickhouse_host`, `your_clickhouse_port`, `your_database_name`, 和 `your_table_name` 为你实际的ClickHouse环境配置。
阅读全文