clickhouse-mysql: error: unrecognized arguments: --src-only-table=airline.ontime

2018-12-04 16:54:15,973/1543942455.973548:DEBUG:class:<class 'clickhouse_mysql.writer.processwriter.ProcessWriter'> process()
2018-12-04 16:54:15,974/1543942455.974890:INFO:CSVWriter() csv_file_path=None csv_file_path_prefix=/root/ch_ csv_file_path_suffix_parts=[] csv_keep_file=False dst_schema=sl dst_table=None
2018-12-04 16:54:15,975/1543942455.975752:INFO:CSVWriter() self.path=/root/ch_1543942455.9753294_c9b08f22-72ec-48f8-8b0a-f82e614de73c.csv
2018-12-04 16:54:15,976/1543942455.976219:WARNING:No events to insert. class: <class 'clickhouse_mysql.writer.csvwriter.CSVWriter'>
2018-12-04 16:54:15,976/1543942455.976954:INFO:CHCSWriter() connection_settings={'host': '127.0.0.1', 'port': 9000, 'user': 'default', 'password': ''} dst_schema=sl dst_table=None
2018-12-04 16:54:15,977/1543942455.977393:DEBUG:class:<class 'clickhouse_mysql.writer.chcsvwriter.CHCSVWriter'> insert 1 rows
--this line was added by me trying to figure out what was going on
2018-12-04 16:54:15,977/1543942455.977737:DEBUG:event: <clickhouse_mysql.event.event.Event object at 0x0000000092b8afe0>, attrs: {'schema': 'sl', 'table': None, 'filename': '/root/ch_1543942455.9753294_c9b08f22-72ec-48f8-8b0a-f82e614de73c.csv', 'fieldnames': None}, events: [<clickhouse_mysql.event.event.Event object at 0x0000000092b8afe0>]
Process Process-11:
Traceback (most recent call last):
  File "/root/pypi/pypy3.5-6.0.0-linux_x86_64-portable/lib-python/3/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/root/pypi/pypy3.5-6.0.0-linux_x86_64-portable/lib-python/3/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/root/pypi/pypy3.5-6.0.0-linux_x86_64-portable/site-packages/clickhouse_mysql-0.0.20180321-py3.5.egg/clickhouse_mysql/writer/processwriter.py", line 34, in process
    writer.push()
  File "/root/pypi/pypy3.5-6.0.0-linux_x86_64-portable/site-packages/clickhouse_mysql-0.0.20180321-py3.5.egg/clickhouse_mysql/writer/csvwriter.py", line 145, in push
    self.next_writer_builder.get().insert(event)
  File "/root/pypi/pypy3.5-6.0.0-linux_x86_64-portable/site-packages/clickhouse_mysql-0.0.20180321-py3.5.egg/clickhouse_mysql/writer/chcsvwriter.py", line 63, in insert
    ', '.join(map(lambda column: '`%s`' % column, event.fieldnames)),
TypeError: map argument #1 must support iteration
2018-12-04 16:54:16,530/1543942456.530018:DEBUG:class:<class 'clickhouse_mysql.writer.processwriter.ProcessWriter'> process() done

# yum install -y clickhouse-mysql
...
--> Finished Dependency Resolution
Error: Package: clickhouse-mysql-0.0.20180319-1.noarch (Altinity_clickhouse)
           Requires: mysql-community-devel
 You could try using --skip-broken to work around the problem
 You could try running: rpm -Va --nofiles --nodigest

# yum install mysql-community-devel
...
No package mysql-community-devel available.
Error: Nothing to do

# yum install mysql-devel
...
Package MariaDB-devel-10.3.13-1.el7.centos.x86_64 already installed and latest version
Nothing to do

1. require Mysql and have sorted primary key.

1. just translate UPDATE event to INSERT, it's friendly for `ReplacingMergeTree`.
2. using `final` ON `SELECT`.

1. just translate DELETE Event to INSERT(always using soft delete).
2. require dst clickhouse table has `deleted_at` field.
3. and also increment `version key`, currently just support `DateTime` field.

1. /usr/bin/clickhouse-server(DB::Exception::Exception(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, int)+0x22) [0x3019952]            
2. /usr/bin/clickhouse-server(DB::Context::getTableImpl(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, DB::Exception*) const+0x1f3) [0x708b523]                                                                                   
3. /usr/bin/clickhouse-server(DB::Context::getTable(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) const+0x73) [0x708b9f3]
4. /usr/bin/clickhouse-server(DB::InterpreterInsertQuery::getTable(DB::ASTInsertQuery const&)+0x191) [0x70edae1]                                                                 
5. /usr/bin/clickhouse-server(DB::InterpreterInsertQuery::execute()+0x63) [0x70ee8c3]
6. /usr/bin/clickhouse-server() [0x71a0727]
7. /usr/bin/clickhouse-server(DB::executeQuery(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, DB::Context&, bool, DB::QueryProcessingStage::Enum)+0x8a) [0x71a245a]
8. /usr/bin/clickhouse-server(DB::TCPHandler::runImpl()+0x4d9) [0x3027179]
9. /usr/bin/clickhouse-server(DB::TCPHandler::run()+0x2b) [0x302836b]
10. /usr/bin/clickhouse-server(Poco::Net::TCPServerConnection::start()+0xf) [0x96107df]
11. /usr/bin/clickhouse-server(Poco::Net::TCPServerDispatcher::run()+0x16a) [0x9610bba]
12. /usr/bin/clickhouse-server(Poco::PooledThread::run()+0x77) [0x972d6b7]
13. /usr/bin/clickhouse-server(Poco::ThreadImpl::runnableEntry(void*)+0x38) [0x9729228]
14. /usr/bin/clickhouse-server() [0x9e659ef]
15. /lib/x86_64-linux-gnu/libpthread.so.0(+0x76db) [0x7f0dbe3286db]
16. /lib/x86_64-linux-gnu/libc.so.6(clone+0x3f) [0x7f0dbd8a788f]


During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/root/clickhouse-mysql-data-reader/clickhouse_mysql/tablemigrator.py", line 194, in migrate_one_table_data                                                               
    self.chwriter.insert(event)
  File "/root/clickhouse-mysql-data-reader/clickhouse_mysql/writer/chwriter.py", line 88, in insert                                                                              
    sys.exit(0)
SystemExit: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/root/clickhouse-mysql-data-reader/clickhouse_mysql/main.py", line 140, in run
    migrator.migrate_all_tables_data()
  File "/root/clickhouse-mysql-data-reader/clickhouse_mysql/tablemigrator.py", line 160, in migrate_all_tables_data                                                              
    self.migrate_one_table_data(db=db, table=table)
  File "/root/clickhouse-mysql-data-reader/clickhouse_mysql/tablemigrator.py", line 200, in migrate_one_table_data                                                               
    self.host,
AttributeError: 'TableMigrator' object has no attribute 'host'
=============
'TableMigrator' object has no attribute 'host'

python3 -m clickhouse_mysql.main \
    --src-server-id=1 \
    --src-wait \
    --nice-pause=1 \
    --src-host=172.31.x.x \
    --src-user=reader \
    --src-password=xx \
    --src-tables=x.x \
    --dst-host=192.168.x.x \
    --dst-schema=chain \
    --dst-table=chain_tx.chain_tx \
    --migrate-table \
    --pump-data \
    --csvpool

            if self.password:
                choptions += " --password='" + self.password + "'"
            bash = "tail -n +2 '{0}' | clickhouse-client {1} --query='{2}'".format(
                event.filename,
                choptions,
                sql,
            )

root@coin-base:~/clickhouse-mysql-data-reader# clickhouse-mysql \
>     --src-server-id=1 \
>     --src-wait \
>     --nice-pause=1 \
>     --src-host=172.31.220.47 \
>     --src-user=reader \
>     --src-password=aaa \
>     --src-tables=chain_tx.chain_tx \
>     --dst-host=127.0.0.1 \
>     --dst-create-table \
>     --migrate-table \
>     --pump-data \
>     --csvpool

Expected=8.  Actual=2.  Position: 187.  Data Length: 189
2020-05-22 05:36:29,351/1590125789.351507:CRITICAL:QUERY FAILED
2020-05-22 05:36:29,351/1590125789.351809:CRITICAL:ex=Code: 53. Type mismatch in VALUES section. Repeat query with types_check=True for detailed info. Column itemid: required argument is not an integer
2020-05-22 05:36:29,351/1590125789.351931:CRITICAL:sql=INSERT INTO `zabbix_i88`.`history_uint` (`itemid`, `clock`, `value`, `ns`) VALUES
2020-05-22 05:37:30,412/1590125850.412596:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:37:30,413/1590125850.413086:WARNING:Result length not requested length:
Expected=8.  Actual=2.  Position: 187.  Data Length: 189
2020-05-22 05:37:31,451/1590125851.451168:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:37:31,451/1590125851.451454:WARNING:Result length not requested length:
Expected=4.  Actual=2.  Position: 8212.  Data Length: 8214
2020-05-22 05:37:32,814/1590125852.814633:CRITICAL:QUERY FAILED
2020-05-22 05:37:32,815/1590125852.815046:CRITICAL:ex=Code: 53. Type mismatch in VALUES section. Repeat query with types_check=True for detailed info. Column itemid: required argument is not an integer
2020-05-22 05:37:32,815/1590125852.815285:CRITICAL:sql=INSERT INTO `zabbix_i88`.`history_uint` (`itemid`, `clock`, `value`, `ns`) VALUES
2020-05-22 05:38:33,900/1590125913.900565:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:38:33,900/1590125913.900909:WARNING:Result length not requested length:
Expected=4.  Actual=2.  Position: 8212.  Data Length: 8214
2020-05-22 05:38:34,910/1590125914.910961:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:38:34,911/1590125914.911283:WARNING:Result length not requested length:
Expected=8.  Actual=1.  Position: 813.  Data Length: 814
2020-05-22 05:38:35,985/1590125915.985317:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:38:35,985/1590125915.985610:WARNING:Result length not requested length:
Expected=8.  Actual=2.  Position: 8212.  Data Length: 8214
2020-05-22 05:38:37,382/1590125917.382241:CRITICAL:QUERY FAILED
2020-05-22 05:38:37,382/1590125917.382560:CRITICAL:ex=Code: 53. Type mismatch in VALUES section. Repeat query with types_check=True for detailed info. Column itemid: required argument is not an integer
2020-05-22 05:38:37,382/1590125917.382715:CRITICAL:sql=INSERT INTO `zabbix_i88`.`history_uint` (`itemid`, `clock`, `value`, `ns`) VALUES
2020-05-22 05:39:38,443/1590125978.443542:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:39:38,443/1590125978.443890:WARNING:Result length not requested length:
Expected=8.  Actual=2.  Position: 8212.  Data Length: 8214
2020-05-22 05:39:39,462/1590125979.462185:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:39:39,462/1590125979.462625:WARNING:Result length not requested length:
Expected=4.  Actual=2.  Position: 812.  Data Length: 814
2020-05-22 05:39:40,536/1590125980.536540:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:39:40,536/1590125980.536929:WARNING:Result length not requested length:
Expected=4.  Actual=2.  Position: 8212.  Data Length: 8214
2020-05-22 05:39:41,932/1590125981.932213:CRITICAL:QUERY FAILED
2020-05-22 05:39:41,932/1590125981.932589:CRITICAL:ex=Code: 53. Type mismatch in VALUES section. Repeat query with types_check=True for detailed info. Column itemid: required argument is not an integer
2020-05-22 05:39:41,932/1590125981.932733:CRITICAL:sql=INSERT INTO `zabbix_i88`.`history_uint` (`itemid`, `clock`, `value`, `ns`) VALUES
2020-05-22 05:40:43,056/1590126043.056870:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:40:43,057/1590126043.057307:WARNING:Result length not requested length:
Expected=4.  Actual=2.  Position: 8212.  Data Length: 8214
2020-05-22 05:40:44,073/1590126044.073156:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:40:44,073/1590126044.073462:WARNING:Result length not requested length:
Expected=4.  Actual=2.  Position: 812.  Data Length: 814
2020-05-22 05:40:45,141/1590126045.141290:WARNING:Got an exception, skip it in blocking mode
2020-05-22 05:40:45,141/1590126045.141573:WARNING:Result length not requested length:
Expected=4.  Actual=2.  Position: 8212.  Data Length: 8214
2020-05-22 05:40:46,519/1590126046.519466:CRITICAL:QUERY FAILED
2020-05-22 05:40:46,519/1590126046.519791:CRITICAL:ex=Code: 53. Type mismatch in VALUES section. Repeat query with types_check=True for detailed info. Column itemid: required argument is not an integer
2020-05-22 05:40:46,519/1590126046.519940:CRITICAL:sql=INSERT INTO `zabbix_i88`.`history_uint` (`itemid`, `clock`, `value`, `ns`) VALUES

/*!80011 SET @@session.default_collation_for_utf8mb4=255*//*!*/;
BEGIN
/*!*/;
# at 90742723
#200522  4:05:37 server id 1  end_log_pos 90742790 CRC32 0xaea15f3c     Table_map: `zabbix_i88`.`history_uint` mapped to number 3568
# at 90742790
#200522  4:05:37 server id 1  end_log_pos 90742978 CRC32 0xebdfe8cd     Write_rows: table id 3568
# at 90742978
#200522  4:05:37 server id 1  end_log_pos 90743116 CRC32 0x563c8236     Write_rows: table id 3568 flags: STMT_END_F
### Extra row info for partitioning: partition: 31
### INSERT INTO `zabbix_i88`.`history_uint`
### SET
###   @1=31109 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @2=1590075509 /* INT meta=0 nullable=0 is_null=0 */
###   @3=2693000000 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @4=603240000 /* INT meta=0 nullable=0 is_null=0 */
### INSERT INTO `zabbix_i88`.`history_uint`
### SET
###   @1=31109 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @2=1590079109 /* INT meta=0 nullable=0 is_null=0 */
###   @3=2693000000 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @4=314724351 /* INT meta=0 nullable=0 is_null=0 */
### INSERT INTO `zabbix_i88`.`history_uint`
### SET
###   @1=31109 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @2=1590082709 /* INT meta=0 nullable=0 is_null=0 */
###   @3=2693000000 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @4=818740007 /* INT meta=0 nullable=0 is_null=0 */
### INSERT INTO `zabbix_i88`.`history_uint`
### SET
###   @1=31109 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @2=1590086309 /* INT meta=0 nullable=0 is_null=0 */
###   @3=2693000000 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @4=254728503 /* INT meta=0 nullable=0 is_null=0 */
### INSERT INTO `zabbix_i88`.`history_uint`
### SET
###   @1=31109 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @2=1590089909 /* INT meta=0 nullable=0 is_null=0 */
###   @3=2693000000 /* LONGINT meta=0 nullable=0 is_null=0 */
###   @4=219237073 /* INT meta=0 nullable=0 is_null=0 */

log_file=/var/log/clickhouse-mysql/main.log
log_level=WARNING
nice_pause=1  #пауза в сек. между попытками чтения binlog
#dry=yes
daemon=False
pid_file=/var/run/clickhouse-client.pid
binlog_position_file=/var/log/clickhouse-mysql/clickhouse-mysql-binlog.pos

mempool=yes
mempool_max_events_num=10000
mempool_max_rows_num=1000
mempool_max_flush_interval=5
csvpool=no
csvpool_file_path_prefix=/tmp/clickhouse-mysql/zbx__
csvpool_keep_files=no

#migrate_table=yes
#Copy existing data from MySQL table(s) with SELECT statement
#Binlog is not read during this procedure - just copy data from the src table(s). 

pump_data=yes
#Use in combination with --src-wait in case would like to continue and wait for new rows 
src_wait=yes
src_resume=yes # продолжить репликацию если задан binlog_position_file

#
# src section
#
src_server_id=1
src_host=172.31.135.38
src_port=3306
src_user="reader"
src_password="**************8"
src_schemas="zabbix_i88"
src_tables=history_log,history,history_str,history_text,history_uint,trends,trends_uint

#src_file=
#src_binlog_file=mysql-bin.000001 #стартовая позиция binlog, если не задан - берется из binlog_position_file
#src_binlog_position=28579794

#
# dst section
#
dst_host=172.*******
dst_port=9000
dst_user="zabbix"
dst_password="***********#"
dst_schema="zabbix_i88"
dst_create_table=no
dst_table_prefix=""

create table history_uint
(
	itemid bigint unsigned not null,
	clock int default 0 not null,
	value bigint unsigned default 0 not null,
	ns int default 0 not null
);
create index history_uint_1
	on history_uint (itemid, clock);

2021-04-27 12:08:01,107/1619507281.107731:DEBUG:class:<class 'clickhouse_mysql.writer.chwriter.CHWriter'> insert 1 event(s)
2021-04-27 12:08:01,481/1619507281.481600:DEBUG:class:<class 'clickhouse_mysql.writer.chwriter.CHWriter'> insert 100000 row(s)
2021-04-27 12:08:01,481/1619507281.481838:DEBUG:schema=vpbx table=incoming_calls self.dst_schema=None self.dst_table=None
2021-04-27 12:08:01,482/1619507281.482273:DEBUG:Connecting. Database: default. User: default
2021-04-27 12:08:01,482/1619507281.482419:DEBUG:Connecting to 127.0.0.1:9000
2021-04-27 12:08:01,484/1619507281.484395:DEBUG:Connected to ClickHouse server version 21.3.4, revision: 54447
2021-04-27 12:08:01,485/1619507281.485070:DEBUG:Query: INSERT INTO `vpbx`.`incoming_calls` (`id`, `client_id`, `date`, `time`, `cids`, `datetime`, `did_num`, `line_status`, `duration`, `billing_minutes`, `Last_App`, `Last_App_ID`, `Last_App_Data`, `call-status`, `dtmf`, `Caller_ID`, `Recd_File_Path`, `agent_name`, `agent_no`, `Lead_Check`) VALUES
2021-04-27 12:08:01,485/1619507281.485458:DEBUG:Block "" send time: 0.000073
2021-04-27 12:08:01,904/1619507281.904045:CRITICAL:QUERY FAILED
2021-04-27 12:08:01,904/1619507281.904250:CRITICAL:ex='NoneType' object has no attribute 'encode'
2021-04-27 12:08:01,904/1619507281.904330:CRITICAL:sql=INSERT INTO `vpbx`.`incoming_calls` (`id`, `client_id`, `date`, `time`, `cids`, `datetime`, `did_num`, `line_status`, `duration`, `billing_minutes`, `Last_App`, `Last_App_ID`, `Last_App_Data`, `call-status`, `dtmf`, `Caller_ID`, `Recd_File_Path`, `agent_name`, `agent_no`, `Lead_Check`) VALUES

VALUES
2021-06-18 22:58:56,737/1624057136.737545:DEBUG:Block "" send time: 0.000042
2021-06-18 22:59:40,788/1624057180.788412:DEBUG:Block "" send time: 37.195755
2021-06-18 22:59:40,798/1624057180.798683:DEBUG:Block "" send time: 0.006171
2021-06-18 23:02:19,345/1624057339.345471:CRITICAL:Critical error: (2013, 'Lost connection to MySQL server during query')
2021-06-18 23:02:19,349/1624057339.349685:CRITICAL:Can not migrate table on db=XXX table=XXX
=============
Traceback (most recent call last):
  File "/home/ec2-user/.local/lib/python3.7/site-packages/clickhouse_mysql/tablemigrator.py", line 203, in migrate_one_table_data
    rows = self.client.cursor.fetchmany(self.pool_max_rows_num)
  File "/home/ec2-user/.local/lib/python3.7/site-packages/MySQLdb/cursors.py", line 435, in fetchmany
    r = self._fetch_row(size or self.arraysize)
  File "/home/ec2-user/.local/lib/python3.7/site-packages/MySQLdb/cursors.py", line 328, in _fetch_row
    return self._result.fetch_row(size, self._fetch_type)
MySQLdb._exceptions.OperationalError: (2013, 'Lost connection to MySQL server during query')

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/ec2-user/.local/lib/python3.7/site-packages/clickhouse_mysql/main.py", line 140, in run
    migrator.migrate_all_tables_data()
  File "/home/ec2-user/.local/lib/python3.7/site-packages/clickhouse_mysql/tablemigrator.py", line 177, in migrate_all_tables_data
    self.migrate_one_table_data(db=db, table=table)
  File "/home/ec2-user/.local/lib/python3.7/site-packages/clickhouse_mysql/tablemigrator.py", line 220, in migrate_one_table_data
    table,
Exception: Can not migrate table on db=ns_dev_data_db table=cdm
=============
Can not migrate table on db=XXX table=XXX
[ec2-user@ip-172-31-19-8 ~]$ 

2019-01-16 15:03:01,284/1547622181.284754:DEBUG:Query: CREATE TABLE IF NOT EXISTS `xxx`.`xxx` (
    `primary_date_field` Date default today(),
    `id` Int32,
    `creator` Nullable(Int32),
    `modifier` Nullable(Int32),
    `gmt_create` DateTime,
    `pay_time` Nullable(DateTime),
    `gmt_modified` DateTime,
    `is_deleted` Nullable(String),
    `shop_id` Int32,
    `order_type` Int32,
    `parent_id` Nullable(Int32),
    `order_sn` String,
    `customer_id` Nullable(Int32),
    `customer_car_id` Int32,
    `expected_time` Nullable(DateTime),
    `goods_amount` Nullable(String),
    `service_amount` Nullable(String),
    `tax_amount` Nullable(String),
    `total_amount` Nullable(String),
    `postscript` Nullable(String),
    `order_status` String,
    `car_license` String,
    `car_brand_id` Nullable(Int32),
    `car_series_id` Nullable(Int32),
    `car_power_id` Nullable(Int32),
    `car_year_id` Nullable(Int32),
    `car_models_id` Nullable(Int32),
    `car_brand` Nullable(String),
    `car_series` Nullable(String),
    `car_power` Nullable(String),
    `car_year` Nullable(String),
    `car_models` Nullable(String),
    `car_company` Nullable(String),
    `import_info` Nullable(String),
    `customer_name` Nullable(String),
    `customer_mobile` Nullable(String),
    `vin` Nullable(String),
    `engine_no` Nullable(String),
    `receiver` Int32,
    `invoice_type` Nullable(Int32),
    `discount` Nullable(String),
    `receiver_name` Nullable(String),
    `operator_name` Nullable(String),
    `order_amount` Nullable(String),
    `pay_status` UInt8,
    `finish_time` Nullable(DateTime),
    `goods_count` Nullable(Int32),
    `service_count` Nullable(Int32),
    `car_alias` Nullable(String),
    `pre_discount_rate` Nullable(String),
    `pre_tax_amount` Nullable(String),
    `pre_preferentia_amount` Nullable(String),
    `pre_coupon_amount` Nullable(String),
    `pre_total_amount` Nullable(String),
    `pay_amount` Nullable(String),
    `sign_amount` Nullable(String),
    `other_insurance_company_id` Nullable(Int32),
    `other_insurance_company_name` Nullable(String),
    `goods_discount` Nullable(String),
    `service_discount` Nullable(String),
    `fee_amount` Nullable(String),
    `fee_discount` Nullable(String),
    `contact_name` Nullable(String),
    `contact_mobile` Nullable(String),
    `insurance_company_id` Nullable(Int32),
    `insurance_company_name` Nullable(String),
    `mileage` Nullable(String),
    `car_color` Nullable(String),
    `buy_time` Nullable(DateTime),
    `customer_address` Nullable(String),
    `oil_meter` Nullable(String),
    `is_notice` Nullable(Int8),
    `is_visit` Nullable(Int32),
    `is_lock` Nullable(String),
    `img_url` Nullable(String),
    `order_tag` UInt8,
    `ver` Nullable(String),
    `refer` UInt8,
    `upkeep_mileage` Nullable(String),
    `create_time` DateTime,
    `proxy_type` Int8,
    `channel_id` Int32,
    `channel_name` String,
    `order_discount_amount` Nullable(String),
    `confirm_time` DateTime,
    `expect_time` Nullable(DateTime),
    `down_payment` String,
    `company` Nullable(String)
) ENGINE = MergeTree(primary_date_field, (id,gmt_create,gmt_modified,order_type,order_sn,customer_car_id,order_status,car_license,receiver,pay_status,confirm_time), 8192)

......

Row 5:
Column 0,   name: buy_time,                     type: Nullable(DateTime), parsed text: ",,??,32,,,?A4369G,C"
ERROR: garbage after Nullable(DateTime): "addy [??],"



tail: write error: Broken pipe
2019-01-16 15:03:01,895/1547622181.895748:DEBUG:class:<class 'clickhouse_mysql.writer.processwriter.ProcessWriter'> process() done


I checked the csv file ,the row 5 is the following text

CREATE TABLE `acct` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `stamp_inserted` datetime NOT NULL,
  `ip_src` char(45) NOT NULL,
  `ip_dst` char(45) NOT NULL,
  `src_port` int(2) unsigned NOT NULL,
  `dst_port` int(2) unsigned NOT NULL,
  `in_bytes` int(10) unsigned NOT NULL,
  `out_bytes` int(10) unsigned NOT NULL,
  `src_host` varchar(1024) DEFAULT NULL,
  `dst_host` varchar(1024) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1391121 DEFAULT CHARSET=latin1

CREATE TABLE default.acct ( id UInt32,  stamp_inserted DateTime,  ip_src String,  ip_dst String,  src_port UInt16,  dst_port UInt16,  in_bytes UInt32,  out_bytes UInt32,  src_host String,  dst_host String) ENGINE = MergeTree PARTITION BY toMonday(stamp_inserted) ORDER BY (stamp_inserted, ip_dst) SETTINGS index_granularity = 8192

python3 -m clickhouse_mysql.main \
    --src-resume \
    --src-wait \
    --nice-pause=1 \
    --log-level=info \
    --log-file=acct.log \
    --src-host=127.0.0.1 \
    --src-user=reader \
    --src-password=qwerty \
    --dst-host=127.0.0.1 \
    --csvpool \
    --csvpool-file-path-prefix=qwe_ \
    --mempool-max-flush-interval=60 \
    --mempool-max-events-num=10000

Process Process-1:
Traceback (most recent call last):
  File "/usr/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/usr/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/root/clickhousr/click-mysql/clickhouse-mysql-data-reader/clickhouse_mysql/writer/processwriter.py", line 30, in process   
    writer.insert(event_or_events)
  File "/root/clickhousr/click-mysql/clickhouse-mysql-data-reader/clickhouse_mysql/writer/csvwriter.py", line 83, in insert
    self.fieldnames = sorted(events[0].column_names())
  File "/root/clickhousr/click-mysql/clickhouse-mysql-data-reader/clickhouse_mysql/event/event.py", line 53, in column_names
    return self.mysql_event.rows[0]['values'].keys()
IndexError: list index out of range
Process Process-2:
Traceback (most recent call last):
  File "/usr/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/usr/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/root/clickhousr/click-mysql/clickhouse-mysql-data-reader/clickhouse_mysql/writer/processwriter.py", line 30, in process
    writer.insert(event_or_events)
  File "/root/clickhousr/click-mysql/clickhouse-mysql-data-reader/clickhouse_mysql/writer/csvwriter.py", line 83, in insert
    self.fieldnames = sorted(events[0].column_names())
  File "/root/clickhousr/click-mysql/clickhouse-mysql-data-reader/clickhouse_mysql/event/event.py", line 53, in column_names
    return self.mysql_event.rows[0]['values'].keys()
IndexError: list index out of range

2017-12-25 13:15:31,936/1514236531.936720:INFO:Starting
2017-12-25 13:15:31,940/1514236531.940504:INFO:raw dbs list len=0
2017-12-25 13:15:31,941/1514236531.941264:INFO:dbs list len=0
2017-12-25 13:15:31,941/1514236531.941383:INFO:raw tables list len=0
2017-12-25 13:15:31,941/1514236531.941494:INFO:tables list len=0
2017-12-25 13:16:24,682/1514236584.682820:INFO:rot now:1514236584.682676 bktttl:1 bktitemsttl: 13 index:t1.acct reason:FLUSH bktsonbelt:2 bktsize:13 beltnum:1
2017-12-25 13:16:44,976/1514236604.976741:INFO:rot now:1514236604.976535 bktttl:2 bktitemsttl: 52 index:t1.acct reason:FLUSH bktsonbelt:2 bktsize:39 beltnum:1
2017-12-25 13:16:44,980/1514236604.980124:INFO:PERF - buckets_per_sec:0.049276 items_per_sec:1.921764 for last 20 sec

/path/to//your/log/from/config { 
    daily
    create 0644 root root
    rotate 7
    compress
    postrotate
        service clickhouse-mysql restart
    endscript
}

2019-01-18 18:01:45,726/1547805705.726506:DEBUG:starting tail -n +2 'repl_d_all_b_1547805705.7249274_70f83868-9d29-46d8-8f21-8338d1fb91d0.csv' | clickhouse-client  --host=127.0.0.1 --port=9000 --user=default --query='INSERT INTO `test`.`legend_order_info` (`buy_time`, `car_alias`, `car_brand`, `car_brand_id`, `car_color`, `car_company`, `car_license`, `car_models`, `car_models_id`, `car_power`, `car_power_id`, `car_series`, `car_series_id`, `car_year`, `car_year_id`, `channel_id`, `channel_name`, `company`, `confirm_time`, `contact_mobile`, `contact_name`, `create_time`, `creator`, `customer_address`, `customer_car_id`, `customer_id`, `customer_mobile`, `customer_name`, `discount`, `down_payment`, `engine_no`, `expect_time`, `expected_time`, `fee_amount`, `fee_discount`, `finish_time`, `gmt_create`, `gmt_modified`, `goods_amount`, `goods_count`, `goods_discount`, `id`, `img_url`, `import_info`, `insurance_company_id`, `insurance_company_name`, `invoice_type`, `is_deleted`, `is_lock`, `is_notice`, `is_visit`, `mileage`, `modifier`, `oil_meter`, `operator_name`, `order_amount`, `order_discount_amount`, `order_sn`, `order_status`, `order_tag`, `order_type`, `other_insurance_company_id`, `other_insurance_company_name`, `parent_id`, `pay_amount`, `pay_status`, `pay_time`, `postscript`, `pre_coupon_amount`, `pre_discount_rate`, `pre_preferentia_amount`, `pre_tax_amount`, `pre_total_amount`, `proxy_type`, `receiver`, `receiver_name`, `refer`, `service_amount`, `service_count`, `service_discount`, `shop_id`, `sign_amount`, `tax_amount`, `total_amount`, `upkeep_mileage`, `ver`, `vin`) FORMAT CSV'
Code: 27. DB::Exception: Cannot parse input: expected , before: foxhch,,,,2019-01-20,0,,0,0,,,,10.13,,,,0.00,0.00,,,,0.00,0,0.00,123945,,,0,,0,N,N,0,0,,0,,,0.00,0.00,,haha,20,0,0,,0,0.00,0,,,0.00,1.00000000,0.00,0.00,0.00,14: (at row 1)

Row 1:
Column 0,   name: buy_time,                     type: Nullable(DateTime),       parsed text: ",,,0,,,,,0,,0,,0,,0"
Column 1,   name: car_alias,                    type: Nullable(String),         parsed text: "13"
Column 2,   name: car_brand,                    type: Nullable(String),         parsed text: "mychannel"
Column 3,   name: car_brand_id,                 type: Nullable(Int32),          parsed text: <EMPTY>
ERROR: garbage after Nullable(Int32): "foxhch,,,,"

# in mysql

root@localhost:test 06:01:21>insert into legend_order_info (id,shop_id,order_status,order_tag,refer,create_time,proxy_type,channel_id,channel_name,down_payment,company) values(123945,123239,'haha',20,11,'2019-01-20',14,13,'mychannel',10.13,'foxhch');
Query OK, 1 row affected (0.01 sec)

# the ddl on clickhouse

t-k8s-a1 :) show create table legend_order_info\G

SHOW CREATE TABLE legend_order_info

Row 1:
──────
statement: CREATE TABLE test.legend_order_info ( id Int32,  creator Nullable(Int32),  modifier Nullable(Int32),  gmt_create Nullable(DateTime),  pay_time Nullable(DateTime),  gmt_modified Nullable(DateTime),  is_deleted Nullable(String),  shop_id Int32,  order_type Nullable(Int32),  parent_id Nullable(Int32),  order_sn Nullable(String),  customer_id Nullable(Int32),  customer_car_id Nullable(Int32),  expected_time Nullable(DateTime),  goods_amount Nullable(Decimal(12, 2)),  service_amount Nullable(Decimal(12, 2)),  tax_amount Nullable(Decimal(12, 2)),  total_amount Nullable(Decimal(12, 2)),  postscript Nullable(String),  order_status String,  car_license Nullable(String),  car_brand_id Nullable(Int32),  car_series_id Nullable(Int32),  car_power_id Nullable(Int32),  car_year_id Nullable(Int32),  car_models_id Nullable(Int32),  car_brand Nullable(String),  car_series Nullable(String),  car_power Nullable(String),  car_year Nullable(String),  car_models Nullable(String),  car_company Nullable(String),  import_info Nullable(String),  customer_name Nullable(String),  customer_mobile Nullable(String),  vin Nullable(String),  engine_no Nullable(String),  receiver Nullable(Int32),  invoice_type Nullable(Int32),  discount Nullable(Decimal(12, 2)),  receiver_name Nullable(String),  operator_name Nullable(String),  order_amount Nullable(Decimal(12, 2)),  pay_status Nullable(UInt8),  finish_time Nullable(DateTime),  goods_count Nullable(Int32),  service_count Nullable(Int32),  car_alias Nullable(String),  pre_discount_rate Nullable(Decimal(12, 2)),  pre_tax_amount Nullable(Decimal(12, 2)),  pre_preferentia_amount Nullable(Decimal(12, 2)),  pre_coupon_amount Nullable(Decimal(12, 2)),  pre_total_amount Nullable(Decimal(12, 2)),  pay_amount Nullable(Decimal(12, 2)),  sign_amount Nullable(Decimal(12, 2)),  other_insurance_company_id Nullable(Int32),  other_insurance_company_name Nullable(String),  goods_discount Nullable(Decimal(12, 2)),  service_discount Nullable(Decimal(12, 2)),  fee_amount Nullable(Decimal(12, 2)),  fee_discount Nullable(Decimal(12, 2)),  contact_name Nullable(String),  contact_mobile Nullable(String),  insurance_company_id Nullable(Int32),  insurance_company_name Nullable(String),  mileage Nullable(String),  car_color Nullable(String),  buy_time Nullable(DateTime),  customer_address Nullable(String),  oil_meter Nullable(String),  is_notice Nullable(Int8),  is_visit Nullable(Int32),  is_lock Nullable(String),  img_url Nullable(String),  order_tag UInt8,  ver Nullable(String),  refer UInt8,  upkeep_mileage Nullable(String),  create_time Date,  order_discount_amount Nullable(Decimal(16, 2)),  confirm_time Nullable(DateTime),  proxy_type UInt8,  channel_id UInt32,  channel_name String,  expect_time Nullable(DateTime),  down_payment Decimal(12, 2),  company Nullable(String)) ENGINE = MergeTree(create_time, (id, create_time), 8192)

1 rows in set. Elapsed: 0.002 sec.