Write from Kafka

Write from Kafka

About Kafka

Apache Kafka is an open-source distributed event streaming platform, used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications. For the key concepts of kafka, please refer to kafka documentation.

kafka topic

Messages in Kafka are organized by topics. A topic may have one or more partitions. We can manage kafka topics through kafka-topics.

create a topic named kafka-events:

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

Alter kafka-events topic to set partitions to 3:

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

Show all topics and partitions in Kafka:

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

Insert into TDengine

We can write data into TDengine via SQL or Schemaless. For more information, please refer to Insert Using SQL or High Performance Writing or Schemaless Writing.

Examples

Python

python Kafka 客户端

For python kafka client, please refer to kafka client. In this document, we use kafka-python.

consume from Kafka

The simple way to consume messages from Kafka is to read messages one by one. The demo is as follows:

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

For higher performance, we can consume message from kafka in batch. The demo is as follows:

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)

multi-threading

For more higher performance we can process data from kafka in multi-thread. We can use python’s ThreadPoolExecutor to achieve multithreading. The demo is as follows:

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

multi-process

For more higher performance, sometimes we use multiprocessing. In this case, the number of Kafka Consumers should not be greater than the number of Kafka Topic Partitions. The demo is as follows:

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()

In addition to python’s built-in multithreading and multiprocessing library, we can also use the third-party library gunicorn.

Examples

1. #! encoding = utf-8
2. import json
3. import time
4. from json import JSONDecodeError
5. from typing import Callable
6. import logging
7. from concurrent.futures import ThreadPoolExecutor, Future

9. import taos
10. from kafka import KafkaConsumer
11. from kafka.consumer.fetcher import ConsumerRecord

14. class Consumer(object):
15. DEFAULT_CONFIGS = {
16. 'kafka_brokers': 'localhost:9092',
17. 'kafka_topic': 'python_kafka',
18. 'kafka_group_id': 'taos',
19. 'taos_host': 'localhost',
20. 'taos_user': 'root',
21. 'taos_password': 'taosdata',
22. 'taos_database': 'power',
23. 'taos_port': 6030,
24. 'timezone': None,
25. 'clean_after_testing': False,
26. 'bath_consume': True,
27. 'batch_size': 1000,
28. 'async_model': True,
29. 'workers': 10
30. }

32. LOCATIONS = ['California.SanFrancisco', 'California.LosAngles', 'California.SanDiego', 'California.SanJose',
33. 'California.PaloAlto', 'California.Campbell', 'California.MountainView', 'California.Sunnyvale',
34. 'California.SantaClara', 'California.Cupertino']

36. CREATE_DATABASE_SQL = 'create database if not exists {} keep 365 duration 10 buffer 16 wal_level 1'
37. USE_DATABASE_SQL = 'use {}'
38. DROP_TABLE_SQL = 'drop table if exists meters'
39. DROP_DATABASE_SQL = 'drop database if exists {}'
40. CREATE_STABLE_SQL = 'create stable meters (ts timestamp, current float, voltage int, phase float) ' \
41. 'tags (location binary(64), groupId int)'
42. CREATE_TABLE_SQL = 'create table if not exists {} using meters tags (\'{}\', {})'
43. INSERT_SQL_HEADER = "insert into "
44. INSERT_PART_SQL = 'power.{} values (\'{}\', {}, {}, {})'

46. def __init__(self, **configs):
47. self.config: dict = self.DEFAULT_CONFIGS
48. self.config.update(configs)
49. self.consumer = KafkaConsumer(
50. self.config.get('kafka_topic'),  # topic
51. bootstrap_servers=self.config.get('kafka_brokers'),
52. group_id=self.config.get('kafka_group_id'),
53. )
54. self.taos = taos.connect(
55. host=self.config.get('taos_host'),
56. user=self.config.get('taos_user'),
57. password=self.config.get('taos_password'),
58. port=self.config.get('taos_port'),
59. timezone=self.config.get('timezone'),
60. )
61. if self.config.get('async_model'):
62. self.pool = ThreadPoolExecutor(max_workers=self.config.get('workers'))
63. self.tasks: list[Future] = []
64. # tags and table mapping # key: {location}_{groupId} value:
65. self.tag_table_mapping = {}
66. i = 0
67. for location in self.LOCATIONS:
68. for j in range(1, 11):
69. table_name = 'd{}'.format(i)
70. self._cache_table(location=location, group_id=j, table_name=table_name)
71. i += 1

73. def init_env(self):
74. # create database and table
75. self.taos.execute(self.DROP_DATABASE_SQL.format(self.config.get('taos_database')))
76. self.taos.execute(self.CREATE_DATABASE_SQL.format(self.config.get('taos_database')))
77. self.taos.execute(self.USE_DATABASE_SQL.format(self.config.get('taos_database')))
78. self.taos.execute(self.DROP_TABLE_SQL)
79. self.taos.execute(self.CREATE_STABLE_SQL)
80. for tags, table_name in self.tag_table_mapping.items():
81. location, group_id = _get_location_and_group(tags)
82. self.taos.execute(self.CREATE_TABLE_SQL.format(table_name, location, group_id))

84. def consume(self):
85. logging.warning('## start consumer topic-[%s]', self.config.get('kafka_topic'))
86. try:
87. if self.config.get('bath_consume'):
88. self._run_batch(self._to_taos_batch)
89. else:
90. self._run(self._to_taos)
91. except KeyboardInterrupt:
92. logging.warning("## caught keyboard interrupt, stopping")
93. finally:
94. self.stop()

96. def stop(self):
97. # close consumer
98. if self.consumer is not None:
99. self.consumer.commit()
100. self.consumer.close()

102. # multi thread
103. if self.config.get('async_model'):
104. for task in self.tasks:
105. while not task.done():
106. pass
107. if self.pool is not None:
108. self.pool.shutdown()

110. # clean data
111. if self.config.get('clean_after_testing'):
112. self.taos.execute(self.DROP_TABLE_SQL)
113. self.taos.execute(self.DROP_DATABASE_SQL.format(self.config.get('taos_database')))
114. # close taos
115. if self.taos is not None:
116. self.taos.close()

118. def _run(self, f: Callable[[ConsumerRecord], bool]):
119. for message in self.consumer:
120. if self.config.get('async_model'):
121. self.pool.submit(f(message))
122. else:
123. f(message)

125. def _run_batch(self, f: Callable[[list[list[ConsumerRecord]]], None]):
126. while True:
127. messages = self.consumer.poll(timeout_ms=500, max_records=self.config.get('batch_size'))
128. if messages:
129. if self.config.get('async_model'):
130. self.pool.submit(f, messages.values())
131. else:
132. f(list(messages.values()))
133. if not messages:
134. time.sleep(0.1)

136. def _to_taos(self, message: ConsumerRecord) -> bool:
137. sql = self.INSERT_SQL_HEADER + self._build_sql(message.value)
138. if len(sql) == 0:  # decode error, skip
139. return True
140. logging.info('## insert sql %s', sql)
141. return self.taos.execute(sql=sql) == 1

143. def _to_taos_batch(self, messages: list[list[ConsumerRecord]]):
144. sql = self._build_sql_batch(messages=messages)
145. if len(sql) == 0:  # decode error, skip
146. return
147. self.taos.execute(sql=sql)

149. def _build_sql(self, msg_value: str) -> str:
150. try:
151. data = json.loads(msg_value)
152. except JSONDecodeError as e:
153. logging.error('## decode message [%s] error ', msg_value, e)
154. return ''
155. location = data.get('location')
156. group_id = data.get('groupId')
157. ts = data.get('ts')
158. current = data.get('current')
159. voltage = data.get('voltage')
160. phase = data.get('phase')

162. table_name = self._get_table_name(location=location, group_id=group_id)
163. return self.INSERT_PART_SQL.format(table_name, ts, current, voltage, phase)

165. def _build_sql_batch(self, messages: list[list[ConsumerRecord]]) -> str:
166. sql_list = []
167. for partition_messages in messages:
168. for message in partition_messages:
169. sql_list.append(self._build_sql(message.value))

171. return self.INSERT_SQL_HEADER + ' '.join(sql_list)

173. def _cache_table(self, location: str, group_id: int, table_name: str):
174. self.tag_table_mapping[_tag_table_mapping_key(location=location, group_id=group_id)] = table_name

176. def _get_table_name(self, location: str, group_id: int) -> str:
177. return self.tag_table_mapping.get(_tag_table_mapping_key(location=location, group_id=group_id))

180. def _tag_table_mapping_key(location: str, group_id: int):
181. return '{}_{}'.format(location, group_id)

184. def _get_location_and_group(key: str) -> (str, int):
185. fields = key.split('_')
186. return fields[0], fields[1]

189. if __name__ == '__main__':
190. consumer = Consumer(async_model=True)
191. consumer.init_env()
192. consumer.consume()

view source code