poyee-data-warehouse/dw_base/datax/plugins/reader/clickhouse_reader.py

# -*- coding:utf-8 -*-

import re
from configparser import ConfigParser

from dw_base.datax.datax_constants import *
from dw_base.datax.plugins.reader.reader import Reader

# ClickHouse reader

CLICK_HOUSE_READER_NAME = 'clickhousereader'
CLICK_HOUSE_READER_PARAMETER_CONNECTION = 'connection'
CLICK_HOUSE_READER_PARAMETER_DATABASE = 'database'
CLICK_HOUSE_READER_PARAMETER_FETCH_SIZE = 'fetchSize'
CLICK_HOUSE_READER_PARAMETER_QUERY_SQL = 'querySql'
CLICK_HOUSE_READER_PARAMETER_TABLE = 'table'
CLICK_HOUSE_READER_PARAMETER_COLUMN = 'column'
CLICK_HOUSE_READER_PARAMETER_WHERE = 'where'
CLICK_HOUSE_READER_PARAMETER_SPLIT_PK = 'splitPk'


class ClickHouseReader(Reader):

    def __init__(self, base_dir: str, config_parser: ConfigParser, start_date: str = None, stop_date: str = None):
        super(ClickHouseReader, self).__init__(base_dir, config_parser, start_date, stop_date)
        self.plugin_name = CLICK_HOUSE_READER_NAME

    def load_others(self):
        start_date = self.start_date
        stop_date = self.stop_date
        database = self.config_parser.get(self.plugin_type, CLICK_HOUSE_READER_PARAMETER_DATABASE)
        self.check_config(CLICK_HOUSE_READER_PARAMETER_DATABASE, database)
        table = self.config_parser.get(self.plugin_type, CLICK_HOUSE_READER_PARAMETER_TABLE)
        self.check_config(CLICK_HOUSE_READER_PARAMETER_TABLE, table)
        fetch_size = self.config_parser.get(self.plugin_type, CLICK_HOUSE_READER_PARAMETER_FETCH_SIZE) or '1000'
        self.parameter[CLICK_HOUSE_READER_PARAMETER_FETCH_SIZE] = fetch_size
        split_pk = self.config_parser.get(self.plugin_type, CLICK_HOUSE_READER_PARAMETER_SPLIT_PK)
        self.parameter[CLICK_HOUSE_READER_PARAMETER_SPLIT_PK] = split_pk
        where = self.config_parser.get(self.plugin_type, CLICK_HOUSE_READER_PARAMETER_WHERE)
        where = where.replace('${start_date}', start_date)
        where = where.replace('${start-date}', start_date)
        where = where.replace('${stop_date}', stop_date)
        where = where.replace('${stop-date}', stop_date)
        self.parameter[CLICK_HOUSE_READER_PARAMETER_WHERE] = where
        jdbc_url: str = self.parameter[DS_CLICK_HOUSE_JDBC_URL]
        matcher = re.search('jdbc:postgresql://(.+?)/(.+)', jdbc_url)
        if matcher:
            if database:
                jdbc_url = jdbc_url.replace(matcher.group(2), database)
        elif jdbc_url.endswith('/'):
            jdbc_url = f'{jdbc_url}{database}'
        else:
            jdbc_url = f'{jdbc_url}/{database}'
        query_sql = self.config_parser.get(self.plugin_type, CLICK_HOUSE_READER_PARAMETER_QUERY_SQL)
        query_sql = query_sql.replace('${start_date}', start_date)
        query_sql = query_sql.replace('${start-date}', start_date)
        query_sql = query_sql.replace('${stop_date}', stop_date)
        query_sql = query_sql.replace('${stop-date}', stop_date)
        if query_sql:
            connection = {
                DS_CLICK_HOUSE_JDBC_URL: jdbc_url.split(','),
                CLICK_HOUSE_READER_PARAMETER_QUERY_SQL: query_sql
            }
        else:
            connection = {
                DS_CLICK_HOUSE_JDBC_URL: jdbc_url.split(','),
                CLICK_HOUSE_READER_PARAMETER_TABLE: table.split(',')
            }
        self.parameter[CLICK_HOUSE_READER_PARAMETER_CONNECTION] = [connection]
        del self.parameter[DS_CLICK_HOUSE_JDBC_URL]

    def load_column(self):
        columns = self.config_parser.get(self.plugin_type, CLICK_HOUSE_READER_PARAMETER_COLUMN).split(',')
        self.parameter[CLICK_HOUSE_READER_PARAMETER_COLUMN] = columns