以下是测试demo，如有错误还请指正，运行错误还望自行解决，谢谢！

python依赖：

sudo apt-get install libmysqlclient-dev
sudo pip install arrow
sudo pip install pykafka
sudo pip install mysql
sudo pip install python-daemon
sudo pip install mysql_connector_python
sudo pip install influxdb
sudo apt-get install libmysqlclient-dev

KafkaProcess.py

#coding=utf-8

import time
import json
import threading
from pykafka import KafkaClient

options = {
    'kafka': {
        'host': '127.0.0.1:9092',
		'topic_name': 'test',
		'consumer_group': 'consumer01',
		'kafka_zookeeper': '127.0.0.1:2181',
		'timeout': 30*1000, #ms
    },
}

class KafkaProcess:
    def __init__(self, options):
        self.kafka_conf = options.get('kafka')
    
    def kafka_consumer_connect(self):
        client = KafkaClient(hosts=self.kafka_conf.get('host'), socket_timeout_ms=self.kafka_conf.get('timeout'), offsets_channel_socket_timeout_ms=self.kafka_conf.get('timeout'))
        topic = client.topics[self.kafka_conf.get('topic_name')]
        consumer = topic.get_balanced_consumer(consumer_group=self.kafka_conf.get('consumer_group'), auto_commit_enable=True, zookeeper_connect=self.kafka_conf.get('kafka_zookeeper'), zookeeper_connection_timeout_ms=self.kafka_conf.get('timeout'))
        return consumer

    def start(self):
        consumer = self.kafka_consumer_connect()
        iterc = consumer.__iter__()
        while True:
            a = None
            try:
                msgx = iterc.next()
                a = json.loads(msgx.value)
                print a
            except Exception as e:
                logging.warning('Exception Kafka Start: {0}'.format(e))
            time.sleep(0.5)
			
if __name__ == '__main__':
	Kafka = KafkaProcess(options)
	Kafka.start()

ESDBProcess.py

#coding=utf-8

import logging
import threading
from Queue import Queue
from elasticsearch import helpers
from elasticsearch import Elasticsearch

options = {
    'debug': True,
    'es': {
        'es_host': '127.0.0.1:9200',
        'index_name': 'test01',
    },
}

class ESDBProcess(threading.Thread):
    def __init__(self, options, es_queue):
        threading.Thread.__init__(self)
        self.es_conf = options.get('es')
        self.data_list = []
        self.es_queue = es_queue
        self.es_conn = Elasticsearch(self.es_conf.get('es_host'))

    def esInsert(self, data):
        try:
            helpers.bulk(self.es_conn, data)
        except Exception as e:
            logging.warning('Exception esInsert: {0}'.format(e))

    def run(self):
        while True:
            if not self.es_queue.empty():
                data = self.es_queue.get()
				if 'test' in data:
					self.data_list.extend(data.values()[0])
					if len(self.data_list) > 100:
						self.esInsert(self.data_list)
						self.data_list = []

if __name__ == '__main__':
	es_queue = Queue()
	ES = ESDBProcess(options, es_queue)
	ES.start()
	i = 0
	while True:
		i += 1
		_list = []
		_dict = {'1':'Hello', '2':'World'}
		_time = arrow.get('2019-08-15 10:32:30')
		sourceD = dict({"time":_time.format('YYYY-MM-DDTHH:mm:ss.000+0800')}, **_dict)
        action = {"_index":options.get('es').get('index_name'), "_type":'Test', "_id":i}
        action['_source'] = sourceD
        _list.append(action)
		es_queue.put({'test':_list})
		time.sleep(1)

InfluxDBProcess.py

#coding=utf-8

import logging
import time
import arrow
import threading
import pandas as pd
from influxdb import InfluxDBClient

options = {
    'influxdb': {
        'server': '127.0.0.1',
        'port': 8086,
        'user': 'admin',
        'password': 'admin',
        'database': 'db_collector'
    },
}

class InfluxDBProcess(threading.Thread):
    def __init__(self, options, influx_queue, recv_queue):
        threading.Thread.__init__(self)
        self.dbconf = options.get('influxdb')
        self.influx_queue = influx_queue
		self.recv_queue = recv_queue
        self.body_list = []
        self.client = InfluxDBClient(self.dbconf.get('server'), self.dbconf.get('port'), self.dbconf.get('user'), self.dbconf.get('password'), self.dbconf.get('database'))

    def influxInsert(self, client):
        if client and self.body_list:
            try:
                logging.info('influxInsert num: %d' % len(self.body_list))
                client.write_points(self.body_list)
                self.body_list = []
            except Exception as e:
                logging.error('Exception influxInsert: {0}'.format(e))
            finally:
                self.body_list = []
				
	def strTimeSubSecond(self, time_str, step):
        try:
            _time_t = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.localtime(int(time.mktime(time.strptime(time_str, "%Y-%m-%dT%H:%M:%SZ"))) - step))
            return _time_t
        except Exception as e:
            pass
			
	def influxSelect(self, reques_interval):
        if self.client:
            try:
                current_time = None
                res = self.client.query('select last(id) as id from Test')
                points = res.get_points()
                for item in points:
                    current_time = item['time']

                sql = ('select id, name, info, time from Test where time >= \'%s\' and time <= \'%s\'' % (self.strTimeSubSecond(current_time, reques_interval), current_time))
                res = self.client.query(sql)
                logging.info('InfluxDB Select last_time=(%s) current_time=(%s)' % (self.last_time, current_time))
                self.last_time = current_time

                pd_temp = pd.DataFrame(res.get_points())
                pd_temp = pd_temp.dropna(subset=['name', 'info'], how='all') #去除两个字段同为NaN的数据
                pd_temp = pd_temp.fillna(0) #将所有NaN替换成0
                self.recv_queue.put({'Test':pd_temp, 'time':current_time})
            except Exception as e:
                logging.error('Exception influxSelect: {0}'.format(e))
				
    def run(self):
        while True:
            if not self.influx_queue.empty():
                data = self.influx_queue.get()
                self.body_list.append(data)
                if len(self.body_list) > 10:
                    self.influxInsert(self.client)

def form_point(self, time_str, amend=int(-28800)):
	try:
		_time = arrow.get(time_str, 'YYYY-MM-DD HH:mm:ss')
		_stamp = int(str(_time.timestamp + amend) + ('0' * 9))
		return _stamp
	except Exception as e:
		pass
			
if __name__ == '__main__':
	influx_queue = Queue()
	recv_queue = Queue()
	Influx = InfluxDBProcess(options, influx_queue, recv_queue)
	Influx.start()
	i = 0
	request_time = 0
	reques_interval = 30
	while True:
		now_time = time.time()
		if now_time - request_time  > reques_interval:
			request_time = now_time
			Influx.influxSelect(reques_interval)
		if not self.recv_queue.empty():
            data = self.recv_queue.get()
			print data
	
		i += 1
		_time = '2019-08-15 10:32:29'
		_dict = {'id':0, 'name':'hanbo', 'info':'txt', 'time':'2019-08-15 10:32:29'}
		_raw.update(time=form_point(_time), tags={'id': i}, fields=_dict, measurement='Test')
        influx_queue.put(_raw)
		time.sleep(1)

MySQLProcess.py

#coding=utf-8

import time
import logging
import threading
import mysql.connector

options = {
    'mysqldb': {
        'server': '127.0.0.1',
        'user': 'root',
        'password': 'root',
        'database': 'db_test'
    }
}

class DatabaseCursor:
    def __init__(self,conn):
        self.cursor = conn.cursor(dictionary = True)

    def __enter__(self):
        return self.cursor

    def __exit__(self,exc_type, exc_value, exc_tb):
        self.cursor.close()
        self.cursor = None

class MySQLProcess(threading.Thread):
    def __init__(self, options, mysql_queue, recv_queue):
        threading.Thread.__init__(self)
        self.dbconf = options.get('mysqldb')
        self.recv_queue = recv_queue
		self.mysql_queue = mysql_queue
        self.conn = mysql.connector.connect(host=self.dbconf.get('server'),user=self.dbconf.get('user'),password=self.dbconf.get('password'),database=self.dbconf.get('database'))
		
	def updateInfoData(self, conn):
        try:
            with DatabaseCursor(conn) as cursor:
                commit_num = 0
                while not self.mysql_queue.empty():
                    data = self.mysql_queue.get()

                    try:
						sql = ('insert into tb_test set id=%d, name=\'%s\', info=\'%s\', time=\'%s\''
						'on duplicate key '
						'update name=\'%s\', info=\'%s\', time=\'%s\'' %
						(data['id'], data['name'], data['info'], data['time'],
						data['name'], data['info'], data['time'])
						) #如果不存在就插入，如果存在就更新
                        cursor.execute(sql)
                    except Exception as e:
                        logging.warning('Exception Update|Insert SQL: {0}'.format(e))
                conn.commit()
        except Exception as e:
            logging.error('Exception updateInfoData: {0}'.format(e))
			
    def selectData(self, conn):
        try:
            with DatabaseCursor(conn) as cursor:
                _list = []
                cursor.execute('select name from tb_test')
                for row in cursor:
                    _list.append(row['name'])
                self.recv_queue.put({'name':_list})
        except Exception as e:
            logging.error('Exception selectData: {0}'.format(e))

    def run(self):
        while True:
            self.selectData(self.conn)
            time.sleep(10)

if __name__ == '__main__':
	mysql_queue = Queue()
	recv_queue = Queue()
	MySQL = MySQLProcess(options, mysql_queue, recv_queue)
	MySQL.start()
	i = 0
	while True:
		if not self.recv_queue.empty():
			data = self.recv_queue.get()
			print data
			
		i += 1
		_dict = {'id':0, 'name':'hanbo', 'info':'txt', 'time':'2019-08-15 10:32:29'}
		_dict['id'] = i
        mysql_queue.put(_dict)
		time.sleep(1)
# python # mysql # kafka
Logo
一门面向 Data 和 AI 的低代码、云原生的开源编程语言

无需安装部署，在线快速体验 Byzer

更多推荐

编译时权限控制

前言 权限控制，对于 MLSQL 而言的重要程度可以说是生命线。 MLSQL 需要面对各式各样的资源访问，比如 MySQL, Oracle,HDFS，Hive，Kafka，Sorl，ElasticSearch，Redis，API，Web等等，不同用户对这些数据源（以及表，列）的权限是不一样的。 传统模式是，每个用户都需要有个 proxy user，然后到每个数据源里面给这个 proxy user

avatar Byzer 白泽

Byzer 支持 JDBC 聚合下推

聚合下推PR链接聚合下推 我们知道 Byzer 支持多数据源和联邦查询，可以方便分析师在一个平台上快速的分析来自多种数据源的数据，从而进行灵活的探索式分析。 使用场景和优势 当前 Byzer 中加载数据的方式会拉取明细数据到 spark 中进行聚合计算，对于小数据量或者在分布式文件系统上的数据源来说是常规操作。但是对于 JDBC 数据源或者有分析能力的 OLAP 系统来说拉取明细数据可能就不是最

avatar Byzer 白泽

Byzer 术语表

Byzer-lang Byzer ，又称为 Byzer-lang，一门面向 Data 和 AI 的低代码、云原生的开源编程语言。 Byzer 是一门结合了声明式编程和命令式编程的混合编程语言，其低代码且类 SQL 的编程逻辑配合内置算法及插件的加持，能帮助数据工作者们高效打通数据链路，完成数据的清洗转换，并快速地进行机器学习相关的训练及预测。 Byzer 希望能够提供一套语言、一个引擎，就能覆盖整

avatar Byzer 白泽