Twisted 또는 Threads를 사용하여 대용량 스트리밍 데이터 처리, Python 대기열

Question

Twitter API 스트리밍 서버에 장기간 연결 한 지저귐 속도가 매우 빨라졌습니다. 나는 무거운 텍스트 처리를 수행하고 내 데이터베이스에 트윗을 저장합니다.

DB에서 텍스트 처리 및 저장을 담당하는 연결 및 콜백 기능에 PyCurl을 사용하고 있습니다. 내 접근법 아래 누가 제대로 작동하지 않는지보십시오.

네트워크 프로그래밍에 익숙하지 않으므로 알고 싶습니다. 이 문제를 해결하기 위해 스레드, 대기열 또는 트위스트 프레임 워크를 어떻게 사용할 수 있습니까?

def process_tweet(): 
    # do some heaving text processing 


def open_stream_connection(): 
    connect = pycurl.Curl() 
    connect.setopt(pycurl.URL, STREAMURL) 
    connect.setopt(pycurl.WRITEFUNCTION, process_tweet) 
    connect.setopt(pycurl.USERPWD, "%s:%s" % (TWITTER_USER, TWITTER_PASS)) 
    connect.perform() 

Answer 1

단일 컴퓨터를 사용하는 것이 좋으면 간단한 설정입니다.

1 개의 스레드가 연결을 허용합니다. 연결이 수락되면 처리를 위해 허용 된 연결을 다른 스레드로 전달합니다.

물론 스레드 대신 프로세스 (예 : multiprocessing 사용)를 사용할 수 있지만, 조언을 제공하기 위해 multiprocessing에 익숙하지 않습니다. 설정은 동일합니다 : 1 프로세스가 연결을 허용 한 다음 하위 프로세스로 전달합니다.

여러 컴퓨터에서 처리를 분할해야하는 경우 간단한 작업은 메시지를 데이터베이스에 채우고 새 레코드에 대해 작업자에게 알리는 것입니다 (이 작업에는 여러 종류의 조정/잠금이 필요합니다. 노동자). 데이터베이스에 충돌하는 것을 피하려면 네트워크 프로세스의 메시지를 작업자에게 파이프해야합니다 (낮은 수준의 네트워킹에는 충분히 익숙하지 않아서 어떻게하는지 알려줄 수 없습니다 :))

Answer 2

나는이 조직을 제안 :

• 하나의 프로세스가 트위터를 읽고 데이터베이스에 트윗을 거즈
• 하나 이상의 프로세스를 데이터베이스를 읽고, 각각을 처리, 새로운 데이터베이스에 삽입합니다. 삭제되거나 처리 된 원본 트윗

즉 두 개의 프로세스/스레드가 더 있습니다. 트윗 데이터베이스는 작업 대기열로 볼 수 있습니다. 여러 작업자 프로세스가 대기열에서 작업 (트윗)을 가져 와서 두 번째 데이터베이스에 데이터를 만듭니다.

Answer 3

메시지가 들어올 때 많은 스레드가 메시지를 수신해야합니다.이 번호는 pycurl을 사용하는 경우 1, httplib를 사용하는 경우 더 높아야합니다. 한 번에 Twitter API에 대해 하나의 쿼리를 처리하므로 처리 할 일정량의 작업이 있습니다.

각 트윗이 도착하면 Queue.Queue로 푸시됩니다. 대기열은 통신에 스레드 안전성을 보장합니다. 각 트윗은 하나의 작업자 스레드에서만 처리됩니다.

작업자 스레드 풀은 대기열에서 읽고 짹짹을 처리합니다. 흥미로운 트윗 만 데이터베이스에 추가해야합니다.

데이터베이스에는 병목 현상이있을 수 있으므로 추가 할 가치가있는 풀의 스레드 수에 제한이 있습니다. 더 많은 스레드가 처리 속도를 높이 지 못하면 더 많은 스레드가 큐를 사용하여 데이터베이스에 액세스하십시오.

이것은 상당히 일반적인 파이썬 관용구입니다.이 아키텍처는 일정 정도만 확장 가능합니다. 즉, 처리 할 수있는 머신은 어느 정도입니다.