不接觸業務不知道更多技術，就無法拓展自己，在推薦領域kafka是必備的技術，那么放馬過來吧，開始，

For Recommendation in Deep learning QQ Second Group 102948747
For Visual in deep learning QQ Group 629530787
I'm here waiting for you
不接受這個網頁的私聊/私信！！
1-安裝kafka

很多庫并不是直接就是這個名字安裝，比如opencv等等，

按照此博文更新pip后再安裝

$ pip install kafka-python

經過測驗發現如下bug，經查是我的分布式redis不支持keys(),把這行注釋即可，

connection.py", line 756, in read_response
    raise response
redis.exceptions.ResponseError: Protocol error: invalid multibulk length

.__conn.keys()

2-消費kafka

class KafkaConsumer(builtins.object)
 |  Consume records from a Kafka cluster.
 |  Keyword Arguments:
 |      bootstrap_servers: 'host[:port]' string (or list of 'host[:port]'
 |          strings) that the consumer should contact to bootstrap initial
 |          cluster metadata. This does not have to be the full node list.
 |          It just needs to have at least one broker that will respond to a
 |          Metadata API Request. Default port is 9092. If no servers are
 |          specified, will default to localhost:9092.
 |      client_id (str): A name for this client. This string is passed in
 |          each request to servers and can be used to identify specific
 |          server-side log entries that correspond to this client. Also
 |          submitted to GroupCoordinator for logging with respect to
 |          consumer group administration. Default: 'kafka-python-{version}'
 |      group_id (str or None): The name of the consumer group to join for dynamic
 |          partition assignment (if enabled), and to use for fetching and
 |          committing offsets. If None, auto-partition assignment (via
 |          group coordinator) and offset commits are disabled.
 |          Default: None

我這里只用第一個和第三個，其他的還有好多上面沒有展示，

消費kafka佇列，因為不需要獲取實時資料，實時的可能會造成性能問題，只需消費kafka佇列即可，如下：主動拉取資料，訂閱主題topic

from kafka import KafkaConsumer
import time

kc = KafkaConsumer(group_id='100000', bootstrap_servers=['10.00.86.11:1001'])
kc.subscribe(topics=('test',))
n=10

>>> while n>0:
...     msg=kc.poll(timeout_ms=120)
...     print(type(msg))
...     n-=1

上面的poll，從指定topic拿到資料

poll(timeout_ms=0, max_records=None, update_offsets=True) method of kafka.consumer.group.KafkaConsumer instance
    Fetch data from assigned topics / partitions.
    
    Records are fetched and returned in batches by topic-partition.
    On each poll, consumer will try to use the last consumed offset as the
    starting offset and fetch sequentially. The last consumed offset can be
    manually set through :meth:`~kafka.KafkaConsumer.seek` or automatically
    set as the last committed offset for the subscribed list of partitions.
    
    Incompatible with iterator interface -- use one or the other, not both.
    
    Arguments:
        timeout_ms (int, optional): Milliseconds spent waiting in poll if
            data is not available in the buffer. If 0, returns immediately
            with any records that are available currently in the buffer,
            else returns empty. Must not be negative. Default: 0
        max_records (int, optional): The maximum number of records returned
            in a single call to :meth:`~kafka.KafkaConsumer.poll`.
            Default: Inherit value from max_poll_records.
    
    Returns:
        dict: Topic to list of records since the last fetch for the
            subscribed list of topics and partitions.

這樣拿到的是key：ConsumerRecord ，value：list，元素是

>>> type(value[0])
<class 'kafka.consumer.fetcher.ConsumerRecord'>

這種難以操作，還是用for遍歷吧，

3-多行程存盤資料

將資料放進佇列中，然后采用多行程進行存盤在redis，由上可知我的是分布式redis，但也可用redis讀取，存盤，洗掉，只是有些函式不能用，比如上面的keys，這種操作是有風險的，

既然是讀取kafka的佇列資訊（或者稱之為資料流），就不可能讀取完后再挨個用管道（pipeline）存盤，肯定是讀取一個存盤一個，這里就采用Queue存盤資料，然后Process同步進行，

可參考我之前的博文，參考1，參考2，參考3.或者自己搜索也可，

愿我們終有重逢之時，而你還記得我們曾經討論的話題，