从 Kafka 写入

Kafka 介绍

Apache Kafka 是开源的分布式消息分发平台，被广泛应用于高性能数据管道、流式数据分析、数据集成和事件驱动类型的应用程序。Kafka 包含 Producer、Consumer 和 Topic，其中 Producer 是向 Kafka 发送消息的进程，Consumer 是从 Kafka 消费消息的进程。Kafka 相关概念可以参考官方文档。

kafka topic

Kafka 的消息按 topic 组织，每个 topic 会有一到多个 partition。可以通过 kafka 的 kafka-topics 管理 topic。

创建名为 kafka-events 的topic:

  1. bin/kafka-topics.sh --create --topic kafka-events --bootstrap-server localhost:9092

修改 kafka-events 的 partition 数量为 3:

  1. bin/kafka-topics.sh --alter --topic kafka-events --partitions 3 --bootstrap-server=localhost:9092

展示所有的 topic 和 partition:

  1. bin/kafka-topics.sh --bootstrap-server=localhost:9092 --describe

写入 TDengine

TDengine 支持 Sql 方式和 Schemaless 方式的数据写入，Sql 方式数据写入可以参考 TDengine SQL 写入 和 TDengine 高效写入。Schemaless 方式数据写入可以参考 TDengine Schemaless 写入 文档。

示例代码

  • Python

python Kafka 客户端

Kafka 的 python 客户端可以参考文档 kafka client。推荐使用 confluent-kafka-python 和 kafka-python。以下示例以 kafka-python 为例。

从 Kafka 消费数据

Kafka 客户端采用 pull 的方式从 Kafka 消费数据，可以采用单条消费的方式或批量消费的方式读取数据。使用 kafka-python 客户端单条消费数据的示例如下：

  1. from kafka import KafkaConsumer
  2. consumer = KafkaConsumer('my_favorite_topic')
  3. for msg in consumer:
  4.      print (msg)

单条消费的方式在数据流量大的情况下往往存在性能瓶颈，导致 Kafka 消息积压，更推荐使用批量消费的方式消费数据。使用 kafka-python 客户端批量消费数据的示例如下：

  1. from kafka import KafkaConsumer
  2. consumer = KafkaConsumer('my_favorite_topic')
  3. while True:
  4.     msgs = consumer.poll(timeout_ms=500, max_records=1000)
  5.     if msgs:
  6.         print (msgs)

Python 多线程

为了提高数据写入效率，通常采用多线程的方式写入数据，可以使用 python 线程池 ThreadPoolExecutor 实现多线程。示例代码如下：

  1. from concurrent.futures import ThreadPoolExecutor, Future
  2. pool = ThreadPoolExecutor(max_workers=10)
  3. pool.submit(...)

Python 多进程

单个python进程不能充分发挥多核 CPU 的性能，有时候我们会选择多进程的方式。在多进程的情况下，需要注意，Kafka Consumer 的数量应该小于等于 Kafka Topic Partition 数量。Python 多进程示例代码如下：

  1. from multiprocessing import Process
  3. ps = []
  4. for i in range(5):
  5.     p = Process(target=Consumer().consume())
  6.     p.start()
  7.     ps.append(p)
  9. for p in ps:
  10.     p.join()

除了 Python 内置的多线程和多进程方式，还可以通过第三方库 gunicorn 实现并发。

完整示例

kafka_example_perform

kafka_example_perform 是示例程序的入口

  1. #! encoding=utf-8
  3. import argparse
  4. import logging
  5. import multiprocessing
  6. import time
  7. from multiprocessing import pool
  9. import kafka_example_common as common
  10. import kafka_example_consumer as consumer
  11. import kafka_example_producer as producer
  13. if __name__ == '__main__':
  14.     parser = argparse.ArgumentParser()
  15.     parser.add_argument('-kafka-broker', type=str, default='localhost:9092',
  16.                         help='kafka borker host. default is `localhost:9200`')
  17.     parser.add_argument('-kafka-topic', type=str, default='tdengine-kafka-practices',
  18.                         help='kafka topic. default is `tdengine-kafka-practices`')
  19.     parser.add_argument('-kafka-group', type=str, default='kafka_practices',
  20.                         help='kafka consumer group. default is `kafka_practices`')
  21.     parser.add_argument('-taos-host', type=str, default='localhost',
  22.                         help='TDengine host. default is `localhost`')
  23.     parser.add_argument('-taos-port', type=int, default=6030, help='TDengine port. default is 6030')
  24.     parser.add_argument('-taos-user', type=str, default='root', help='TDengine username, default is `root`')
  25.     parser.add_argument('-taos-password', type=str, default='taosdata', help='TDengine password, default is `taosdata`')
  26.     parser.add_argument('-taos-db', type=str, default='tdengine_kafka_practices',
  27.                         help='TDengine db name, default is `tdengine_kafka_practices`')
  28.     parser.add_argument('-table-count', type=int, default=100, help='TDengine sub-table count, default is 100')
  29.     parser.add_argument('-table-items', type=int, default=1000, help='items in per sub-tables, default is 1000')
  30.     parser.add_argument('-message-type', type=str, default='line',
  31.                         help='kafka message type. `line` or `json`. default is `line`')
  32.     parser.add_argument('-max-poll', type=int, default=1000, help='max poll for kafka consumer')
  33.     parser.add_argument('-threads', type=int, default=10, help='thread count for deal message')
  34.     parser.add_argument('-processes', type=int, default=1, help='process count')
  36.     args = parser.parse_args()
  37.     total = args.table_count * args.table_items
  39.     logging.warning("## start to prepare testing data...")
  40.     prepare_data_start = time.time()
  41.     producer.produce_total(100, args.kafka_broker, args.kafka_topic, args.message_type, total, args.table_count)
  42.     prepare_data_end = time.time()
  43.     logging.warning("## prepare testing data finished! spend-[%s]", prepare_data_end - prepare_data_start)
  45.     logging.warning("## start to create database and tables ...")
  46.     create_db_start = time.time()
  47.     # create database and table
  48.     common.create_database_and_tables(host=args.taos_host, port=args.taos_port, user=args.taos_user,
  49.                                       password=args.taos_password, db=args.taos_db, table_count=args.table_count)
  50.     create_db_end = time.time()
  51.     logging.warning("## create database and tables finished! spend [%s]", create_db_end - create_db_start)
  53.     processes = args.processes
  55.     logging.warning("## start to consume data and insert into TDengine...")
  56.     consume_start = time.time()
  57.     if processes > 1:  # multiprocess
  58.         multiprocessing.set_start_method("spawn")
  59.         pool = pool.Pool(processes)
  61.         consume_start = time.time()
  62.         for _ in range(processes):
  63.             pool.apply_async(func=consumer.consume, args=(
  64.                 args.kafka_broker, args.kafka_topic, args.kafka_group, args.taos_host, args.taos_port, args.taos_user,
  65.                 args.taos_password, args.taos_db, args.message_type, args.max_poll, args.threads))
  66.         pool.close()
  67.         pool.join()
  68.     else:
  69.         consume_start = time.time()
  70.         consumer.consume(kafka_brokers=args.kafka_broker, kafka_topic=args.kafka_topic, kafka_group_id=args.kafka_group,
  71.                          taos_host=args.taos_host, taos_port=args.taos_port, taos_user=args.taos_user,
  72.                          taos_password=args.taos_password, taos_database=args.taos_db, message_type=args.message_type,
  73.                          max_poll=args.max_poll, workers=args.threads)
  74.     consume_end = time.time()
  75.     logging.warning("## consume data and insert into TDengine over! spend-[%s]", consume_end - consume_start)
  77.     # print report
  78.     logging.warning(
  79.         "\n#######################\n"
  80.         "     Prepare data      \n"
  81.         "#######################\n"
  82.         "# data_type # %s  \n"
  83.         "# total     # %s  \n"
  84.         "# spend     # %s s\n"
  85.         "#######################\n"
  86.         "     Create database   \n"
  87.         "#######################\n"
  88.         "# stable    # 1  \n"
  89.         "# sub-table # 100  \n"
  90.         "# spend     # %s s \n"
  91.         "#######################\n"
  92.         "        Consume        \n"
  93.         "#######################\n"
  94.         "# data_type   # %s  \n"
  95.         "# threads     # %s  \n"
  96.         "# processes   # %s  \n"
  97.         "# total_count # %s  \n"
  98.         "# spend       # %s s\n"
  99.         "# per_second  # %s  \n"
  100.         "#######################\n",
  101.         args.message_type, total, prepare_data_end - prepare_data_start, create_db_end - create_db_start,
  102.         args.message_type, args.threads, processes, total, consume_end - consume_start,
  103.                                   total / (consume_end - consume_start))

查看源码

kafka_example_common

kafka_example_common 是示例程序的公共代码

  1. #! encoding = utf-8
  2. import taos
  4. LOCATIONS = ['California.SanFrancisco', 'California.LosAngles', 'California.SanDiego', 'California.SanJose',
  5.              'California.PaloAlto', 'California.Campbell', 'California.MountainView', 'California.Sunnyvale',
  6.              'California.SantaClara', 'California.Cupertino']
  8. CREATE_DATABASE_SQL = 'create database if not exists {} keep 365 duration 10 buffer 16 wal_level 1'
  9. USE_DATABASE_SQL = 'use {}'
  10. DROP_TABLE_SQL = 'drop table if exists meters'
  11. DROP_DATABASE_SQL = 'drop database if exists {}'
  12. CREATE_STABLE_SQL = 'create stable meters (ts timestamp, current float, voltage int, phase float) tags ' \
  13.                     '(location binary(64), groupId int)'
  14. CREATE_TABLE_SQL = 'create table if not exists {} using meters tags (\'{}\', {})'
  17. def create_database_and_tables(host, port, user, password, db, table_count):
  18.     tags_tables = _init_tags_table_names(table_count=table_count)
  19.     conn = taos.connect(host=host, port=port, user=user, password=password)
  21.     conn.execute(DROP_DATABASE_SQL.format(db))
  22.     conn.execute(CREATE_DATABASE_SQL.format(db))
  23.     conn.execute(USE_DATABASE_SQL.format(db))
  24.     conn.execute(DROP_TABLE_SQL)
  25.     conn.execute(CREATE_STABLE_SQL)
  26.     for tags in tags_tables:
  27.         location, group_id = _get_location_and_group(tags)
  28.         tables = tags_tables[tags]
  29.         for table_name in tables:
  30.             conn.execute(CREATE_TABLE_SQL.format(table_name, location, group_id))
  31.     conn.close()
  34. def clean(host, port, user, password, db):
  35.     conn = taos.connect(host=host, port=port, user=user, password=password)
  36.     conn.execute(DROP_DATABASE_SQL.format(db))
  37.     conn.close()
  40. def _init_tags_table_names(table_count):
  41.     tags_table_names = {}
  42.     group_id = 0
  43.     for i in range(table_count):
  44.         table_name = 'd{}'.format(i)
  45.         location_idx = i % len(LOCATIONS)
  46.         location = LOCATIONS[location_idx]
  47.         if location_idx == 0:
  48.             group_id += 1
  49.             if group_id > 10:
  50.                 group_id -= 10
  51.         key = _tag_table_mapping_key(location=location, group_id=group_id)
  52.         if key not in tags_table_names:
  53.             tags_table_names[key] = []
  54.         tags_table_names[key].append(table_name)
  56.     return tags_table_names
  59. def _tag_table_mapping_key(location, group_id):
  60.     return '{}_{}'.format(location, group_id)
  63. def _get_location_and_group(key):
  64.     fields = key.split('_')
  65.     return fields[0], fields[1]

查看源码

kafka_example_producer

kafka_example_producer 是示例程序的 producer 代码，负责生成并发送测试数据到 kafka

  1. #! encoding = utf-8
  2. import json
  3. import random
  4. import threading
  5. from concurrent.futures import ThreadPoolExecutor, Future
  6. from datetime import datetime
  8. from kafka import KafkaProducer
  10. locations = ['California.SanFrancisco', 'California.LosAngles', 'California.SanDiego', 'California.SanJose',
  11.              'California.PaloAlto', 'California.Campbell', 'California.MountainView', 'California.Sunnyvale',
  12.              'California.SantaClara', 'California.Cupertino']
  14. producers: list[KafkaProducer] = []
  16. lock = threading.Lock()
  17. start = 1640966400
  20. def produce_total(workers, broker, topic, message_type, total, table_count):
  21.     if len(producers) == 0:
  22.         lock.acquire()
  23.         if len(producers) == 0:
  24.             _init_kafka_producers(broker=broker, count=10)
  25.         lock.release()
  26.     pool = ThreadPoolExecutor(max_workers=workers)
  27.     futures = []
  28.     for _ in range(0, workers):
  29.         futures.append(pool.submit(_produce_total, topic, message_type, int(total / workers), table_count))
  30.     pool.shutdown()
  31.     for f in futures:
  32.         f.result()
  33.     _close_kafka_producers()
  36. def _produce_total(topic, message_type, total, table_count):
  37.     producer = _get_kafka_producer()
  38.     for _ in range(total):
  39.         message = _get_fake_date(message_type=message_type, table_count=table_count)
  40.         producer.send(topic=topic, value=message.encode(encoding='utf-8'))
  43. def _init_kafka_producers(broker, count):
  44.     for _ in range(count):
  45.         p = KafkaProducer(bootstrap_servers=broker, batch_size=64 * 1024, linger_ms=300, acks=0)
  46.         producers.append(p)
  49. def _close_kafka_producers():
  50.     for p in producers:
  51.         p.close()
  54. def _get_kafka_producer():
  55.     return producers[random.randint(0, len(producers) - 1)]
  58. def _get_fake_date(table_count, message_type='json'):
  59.     if message_type == 'json':
  60.         return _get_json_message(table_count=table_count)
  61.     if message_type == 'line':
  62.         return _get_line_message(table_count=table_count)
  63.     return ''
  66. def _get_json_message(table_count):
  67.     return json.dumps({
  68.         'ts': _get_timestamp(),
  69.         'current': random.randint(0, 1000) / 100,
  70.         'voltage': random.randint(105, 115),
  71.         'phase': random.randint(0, 32000) / 100000,
  72.         'location': random.choice(locations),
  73.         'groupId': random.randint(1, 10),
  74.         'table_name': _random_table_name(table_count)
  75.     })
  78. def _get_line_message(table_count):
  79.     return "{} values('{}', {}, {}, {})".format(
  80.         _random_table_name(table_count),  # table
  81.         _get_timestamp(),  # ts
  82.         random.randint(0, 1000) / 100,  # current
  83.         random.randint(105, 115),  # voltage
  84.         random.randint(0, 32000) / 100000,  # phase
  85.     )
  88. def _random_table_name(table_count):
  89.     return 'd{}'.format(random.randint(0, table_count - 1))
  92. def _get_timestamp():
  93.     global start
  94.     lock.acquire(blocking=True)
  95.     start += 0.001
  96.     lock.release()
  97.     return datetime.fromtimestamp(start).strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]

查看源码

kafka_example_consumer

kafka_example_consumer 是示例程序的 consumer 代码，负责从 kafka 消费数据，并写入到 TDengine

  1. #! encoding = utf-8
  2. import json
  3. import logging
  4. import time
  5. from concurrent.futures import ThreadPoolExecutor, Future
  6. from json import JSONDecodeError
  7. from typing import Callable
  9. import taos
  10. from kafka import KafkaConsumer
  11. from kafka.consumer.fetcher import ConsumerRecord
  13. import kafka_example_common as common
  16. class Consumer(object):
  17.     DEFAULT_CONFIGS = {
  18.         'kafka_brokers': 'localhost:9092',  # kafka broker
  19.         'kafka_topic': 'tdengine_kafka_practices',
  20.         'kafka_group_id': 'taos',
  21.         'taos_host': 'localhost',  # TDengine host
  22.         'taos_port': 6030,  # TDengine port
  23.         'taos_user': 'root',  # TDengine user name
  24.         'taos_password': 'taosdata',  # TDengine password
  25.         'taos_database': 'power',  # TDengine database
  26.         'message_type': 'json',  # message format, 'json' or 'line'
  27.         'clean_after_testing': False,  # if drop database after testing
  28.         'max_poll': 1000,  # poll size for batch mode
  29.         'workers': 10,  # thread count for multi-threading
  30.         'testing': False
  31.     }
  33.     INSERT_SQL_HEADER = "insert into "
  34.     INSERT_PART_SQL = '{} values (\'{}\', {}, {}, {})'
  36.     def __init__(self, **configs):
  37.         self.config = self.DEFAULT_CONFIGS
  38.         self.config.update(configs)
  40.         self.consumer = None
  41.         if not self.config.get('testing'):
  42.             self.consumer = KafkaConsumer(
  43.                 self.config.get('kafka_topic'),
  44.                 bootstrap_servers=self.config.get('kafka_brokers'),
  45.                 group_id=self.config.get('kafka_group_id'),
  46.             )
  48.         self.conns = taos.connect(
  49.             host=self.config.get('taos_host'),
  50.             port=self.config.get('taos_port'),
  51.             user=self.config.get('taos_user'),
  52.             password=self.config.get('taos_password'),
  53.             db=self.config.get('taos_database'),
  54.         )
  55.         if self.config.get('workers') > 1:
  56.             self.pool = ThreadPoolExecutor(max_workers=self.config.get('workers'))
  57.             self.tasks = []
  58.         # tags and table mapping # key: {location}_{groupId} value:
  60.     def consume(self):
  61.         """
  63.         consume data from kafka and deal. Base on `message_type`, `bath_consume`, `insert_by_table`,
  64.         there are several deal function.
  65.         :return:
  66.         """
  67.         self.conns.execute(common.USE_DATABASE_SQL.format(self.config.get('taos_database')))
  68.         try:
  69.             if self.config.get('message_type') == 'line':  # line
  70.                 self._run(self._line_to_taos)
  71.             if self.config.get('message_type') == 'json':  # json
  72.                 self._run(self._json_to_taos)
  73.         except KeyboardInterrupt:
  74.             logging.warning("## caught keyboard interrupt, stopping")
  75.         finally:
  76.             self.stop()
  78.     def stop(self):
  79.         """
  81.         stop consuming
  82.         :return:
  83.         """
  84.         # close consumer
  85.         if self.consumer is not None:
  86.             self.consumer.commit()
  87.             self.consumer.close()
  89.         # multi thread
  90.         if self.config.get('workers') > 1:
  91.             if self.pool is not None:
  92.                 self.pool.shutdown()
  93.             for task in self.tasks:
  94.                 while not task.done():
  95.                     time.sleep(0.01)
  97.         # clean data
  98.         if self.config.get('clean_after_testing'):
  99.             self.conns.execute(common.DROP_TABLE_SQL)
  100.             self.conns.execute(common.DROP_DATABASE_SQL.format(self.config.get('taos_database')))
  101.         # close taos
  102.         if self.conns is not None:
  103.             self.conns.close()
  105.     def _run(self, f):
  106.         """
  108.         run in batch consuming mode
  109.         :param f:
  110.         :return:
  111.         """
  112.         i = 0  # just for test.
  113.         while True:
  114.             messages = self.consumer.poll(timeout_ms=100, max_records=self.config.get('max_poll'))
  115.             if messages:
  116.                 if self.config.get('workers') > 1:
  117.                     self.pool.submit(f, messages.values())
  118.                 else:
  119.                     f(list(messages.values()))
  120.             if not messages:
  121.                 i += 1  # just for test.
  122.                 time.sleep(0.1)
  123.             if i > 3:  # just for test.
  124.                 logging.warning('## test over.')  # just for test.
  125.                 return  # just for test.
  127.     def _json_to_taos(self, messages):
  128.         """
  130.         convert a batch of json data to sql, and insert into TDengine
  131.         :param messages:
  132.         :return:
  133.         """
  134.         sql = self._build_sql_from_json(messages=messages)
  135.         self.conns.execute(sql=sql)
  137.     def _line_to_taos(self, messages):
  138.         """
  140.         convert a batch of lines data to sql, and insert into TDengine
  141.         :param messages:
  142.         :return:
  143.         """
  144.         lines = []
  145.         for partition_messages in messages:
  146.             for message in partition_messages:
  147.                 lines.append(message.value.decode())
  148.         sql = self.INSERT_SQL_HEADER + ' '.join(lines)
  149.         self.conns.execute(sql=sql)
  151.     def _build_single_sql_from_json(self, msg_value):
  152.         try:
  153.             data = json.loads(msg_value)
  154.         except JSONDecodeError as e:
  155.             logging.error('## decode message [%s] error ', msg_value, e)
  156.             return ''
  157.         # location = data.get('location')
  158.         # group_id = data.get('groupId')
  159.         ts = data.get('ts')
  160.         current = data.get('current')
  161.         voltage = data.get('voltage')
  162.         phase = data.get('phase')
  163.         table_name = data.get('table_name')
  165.         return self.INSERT_PART_SQL.format(table_name, ts, current, voltage, phase)
  167.     def _build_sql_from_json(self, messages):
  168.         sql_list = []
  169.         for partition_messages in messages:
  170.             for message in partition_messages:
  171.                 sql_list.append(self._build_single_sql_from_json(message.value))
  172.         return self.INSERT_SQL_HEADER + ' '.join(sql_list)
  175. def test_json_to_taos(consumer: Consumer):
  176.     records = [
  177.         [
  178.             ConsumerRecord(checksum=None, headers=None, offset=1, key=None,
  179.                            value=json.dumps({'table_name': 'd0',
  180.                                              'ts': '2022-12-06 15:13:38.643',
  181.                                              'current': 3.41,
  182.                                              'voltage': 105,
  183.                                              'phase': 0.02027, }),
  184.                            partition=1, topic='test', serialized_key_size=None, serialized_header_size=None,
  185.                            serialized_value_size=None, timestamp=time.time(), timestamp_type=None),
  186.             ConsumerRecord(checksum=None, headers=None, offset=1, key=None,
  187.                            value=json.dumps({'table_name': 'd1',
  188.                                              'ts': '2022-12-06 15:13:39.643',
  189.                                              'current': 3.41,
  190.                                              'voltage': 102,
  191.                                              'phase': 0.02027, }),
  192.                            partition=1, topic='test', serialized_key_size=None, serialized_header_size=None,
  193.                            serialized_value_size=None, timestamp=time.time(), timestamp_type=None),
  194.         ]
  195.     ]
  197.     consumer._json_to_taos(messages=records)
  200. def test_line_to_taos(consumer: Consumer):
  201.     records = [
  202.         [
  203.             ConsumerRecord(checksum=None, headers=None, offset=1, key=None,
  204.                            value="d0 values('2023-01-01 00:00:00.001', 3.49, 109, 0.02737)".encode('utf-8'),
  205.                            partition=1, topic='test', serialized_key_size=None, serialized_header_size=None,
  206.                            serialized_value_size=None, timestamp=time.time(), timestamp_type=None),
  207.             ConsumerRecord(checksum=None, headers=None, offset=1, key=None,
  208.                            value="d1 values('2023-01-01 00:00:00.002', 6.19, 112, 0.09171)".encode('utf-8'),
  209.                            partition=1, topic='test', serialized_key_size=None, serialized_header_size=None,
  210.                            serialized_value_size=None, timestamp=time.time(), timestamp_type=None),
  211.         ]
  212.     ]
  213.     consumer._line_to_taos(messages=records)
  216. def consume(kafka_brokers, kafka_topic, kafka_group_id, taos_host, taos_port, taos_user,
  217.             taos_password, taos_database, message_type, max_poll, workers):
  218.     c = Consumer(kafka_brokers=kafka_brokers, kafka_topic=kafka_topic, kafka_group_id=kafka_group_id,
  219.                  taos_host=taos_host, taos_port=taos_port, taos_user=taos_user, taos_password=taos_password,
  220.                  taos_database=taos_database, message_type=message_type, max_poll=max_poll, workers=workers)
  221.     c.consume()
  224. if __name__ == '__main__':
  225.     consumer = Consumer(testing=True)
  226.     common.create_database_and_tables(host='localhost', port=6030, user='root', password='taosdata', db='py_kafka_test',
  227.                                       table_count=10)
  228.     consumer.conns.execute(common.USE_DATABASE_SQL.format('py_kafka_test'))
  229.     test_json_to_taos(consumer)
  230.     test_line_to_taos(consumer)
  231.     common.clean(host='localhost', port=6030, user='root', password='taosdata', db='py_kafka_test')

查看源码

执行步骤

执行 Python 示例程序

  1. 1. 安装并启动 kafka
  3. 2. python 环境准备
  4.     - 安装 python3
  5.     - 安装 taospy
  6.     - 安装 kafka-python
  8. 3. 执行示例程序
  10. 程序的执行入口是 `kafka_example_perform.py`，获取程序完整的执行参数，请执行 help 命令。
  12. ```
  13. python3 kafka_example_perform.py --help
  14. ```
  16. 以下为创建 100 个子表，每个子表 20000 条数据，kafka max poll 为 100，一个进程，每个进程一个处理线程的程序执行命令
  18. ```
  19. python3 kafka_example_perform.py -table-count=100 -table-items=20000 -max-poll=100 -threads=1 -processes=1
  20. ```