使用 DataFrame 更改我的 Clickhouse 数据库表

Question

我想要的是我有一个数据框（click_df2）：-

        date  L120_active_cohort_logins  L120_active_cohort  percentage_L120_active_cohort_logins
0 2022-09-03                      45000              199000                             22.621906
1 2022-09-04                      40000              200000                             19.092138

现在基于这个 DataFrame 我想根据 DataFrame 中给出的日期更改所有列的值

这就是我创建 clickhouse 表的方式：-

query = '''CREATE TABLE IF NOT EXISTS repeat_day_by_last_120_active_cohort_v1
    (
        date Date,
        L120_active_cohort_logins Int,
        L120_active_cohort Int,
        percentage_L120_active_cohort_logins Float
    ) ENGINE = MergeTree() 
    ORDER BY date'''

代码如下这是我想要做的： -

    click_df2 = pd.read_csv(f'{location}/csv_files/main_data.csv',header=0)
    click_df2['date'] = pd.to_datetime(click_df2['date'],dayfirst=True)
    client.execute(f'''ALTER TABLE repeat_day_by_last_120_active_cohort_v1 \
    UPDATE 'L120_active_cohort_logins' = "{click_df2["L120_active_cohort_logins"]}", \
    'L120_active_cohort' = "{click_df2["L120_active_cohort"]}", \
    'percentage_L120_active_cohort_logins' = "{click_df2["percentage_L120_active_cohort_logins"]}" \
    WHERE 'date' = "{click_df2["date"]}"''')

clickhouse 表中存在的数据 repeat_day_by_last_120_active_cohort_v1 ：-

        date  L120_active_cohort_logins  L120_active_cohort  percentage_L120_active_cohort_logins
0 2022-09-01                      32679              195345                             16.728865
1 2022-09-02                      32938              196457                             16.766010
2 2022-09-03                      40746              197586                             20.621906
3 2022-09-04                      33979              198799                             17.092138

更改表 repeat_day_by_last_120_active_cohort_v1 数据后应为：-

        date  L120_active_cohort_logins  L120_active_cohort  percentage_L120_active_cohort_logins
0 2022-09-01                      32679              195345                             16.728865
1 2022-09-02                      32938              196457                             16.766010
2 2022-09-03                      45000              199000                             22.621906
3 2022-09-04                      40000              200000                             19.092138

Answer 1

只需使用to_dict('records') 在循环内运行突变。这是一个例子：

# docker-compose.yaml
version: "3"
services:
  click_server:
   image: yandex/clickhouse-server
   expose:
     - "8123"
   ports:
     - "8123:8123"
     - "9000:9000"
     - "9009:9009"

运行clickhouse 容器：docker-compose up -d。创建一个.py 脚本：

import pandas as pd
from clickhouse_driver import Client


def print_clickhouse_data(client_: Client):
    print('clickhouse data:')
    print(client_.query_dataframe("""
        SELECT date,
               L120_active_cohort_logins AS logins,
               L120_active_cohort AS cohorts,
               percentage_L120_active_cohort_logins AS percent
          FROM repeat_day_by_last_120_active_cohort_v1
    """))


client = Client(host='localhost')
client.execute('DROP TABLE IF EXISTS repeat_day_by_last_120_active_cohort_v1;')
client.execute("""
CREATE TABLE IF NOT EXISTS repeat_day_by_last_120_active_cohort_v1
    (
        date Date,
        L120_active_cohort_logins Int64,
        L120_active_cohort Int64,
        percentage_L120_active_cohort_logins Float64
    ) ENGINE = MergeTree() 
    ORDER BY date
""")


# init clickhouse data
client.insert_dataframe(
    'INSERT INTO repeat_day_by_last_120_active_cohort_v1 VALUES',
    pd.DataFrame({
        'date': ['2022-09-03', '2022-09-04'],
        'L120_active_cohort_logins': [40746, 33979],
        'L120_active_cohort': [197586, 198799],
        'percentage_L120_active_cohort_logins': [20.621906, 17.092138],
    }),
    settings=dict(use_numpy=True),
)


print_clickhouse_data(client)
# your df for update(in your case from csv...)
click_df = pd.DataFrame({
    'date': ['2022-09-03', '2022-09-04'],
    'L120_active_cohort_logins': [45000, 40000],
    'L120_active_cohort': [199000, 200000],
    'percentage_L120_active_cohort_logins': [22.621906, 19.092138],
})


# update data
for line in click_df.to_dict('records'):  # type: dict
    client.execute(
        """
        ALTER TABLE repeat_day_by_last_120_active_cohort_v1
              UPDATE L120_active_cohort_logins = %(logins)s,
                     L120_active_cohort = %(cohort)s,
                     percentage_L120_active_cohort_logins = %(percent)s
        WHERE date = %(date)s
        """,
        params=dict(
            logins=line['L120_active_cohort_logins'],
            cohort=line['L120_active_cohort'],
            percent=line['percentage_L120_active_cohort_logins'],
            date=line['date'],
        ),
        settings=dict(mutations_sync=2),
    )

    print('{} updated'.format(line['date']))


print_clickhouse_data(client)

运行脚本：

clickhouse data:
         date  logins  cohorts    percent
0  2022-09-03   40746   197586  20.621906
1  2022-09-04   33979   198799  17.092138
2022-09-03 updated
2022-09-04 updated
clickhouse data:
         date  logins  cohorts    percent
0  2022-09-03   45000   199000  22.621906
1  2022-09-04   40000   200000  19.092138

请参阅mutations_sync 设置