我正在從 twitter API 獲取推文，然后通過 tcp 連接將它們轉發到 spark 正在從中讀取資料的套接字中。這是我的代碼
供參考line看起來像這樣
{
data : {
text: "some tweet",
id: some number
}
matching_rules: [{tag: "some string", id: same number}, {tag:...}]
}
def ingest_into_spark(tcp_conn, stream):
for line in stream.iter_lines():
if not (line is None):
try :
# print(line)
tweet = json.loads(line)['matching_rules'][0]['tag']
# tweet = json.loads(line)['data']['text']
print(tweet, type(tweet), len(tweet))
tcp_conn.sendall(tweet.encode('utf-8'))
except Exception as e:
print("Exception in ingesting data: ", e)
火花側代碼：
print(f"Connecting to {SPARK_IP}:{SPARK_PORT}...")
input_stream = streaming_context.socketTextStream(SPARK_IP, int(SPARK_PORT))
print(f"Connected to {SPARK_IP}:{SPARK_PORT}")
tags = input_stream.flatMap(lambda tags: tags.strip().split())
mapped_hashtags = tags.map(lambda hashtag: (hashtag, 1))
counts=mapped_hashtags.reduceByKey(lambda a, b: a b)
counts.pprint()
無論我做什么，spark 都不會讀取通過流發送的資料。但是當我用這條線替換這條線tweet = json.loads(line)['matching_rules'][0]['tag']時，tweet = json.loads(line)['data']['text']它突然按預期作業。我嘗試在兩行中列印推文的內容及其型別，并在兩行中??列印其字串。唯一的區別是第一個有實際的推文，而第二個只有 1 個單詞。
我嘗試了許多不同型別的輸入并對輸入進行硬編碼。但我無法想象為什么讀取 json 的不同欄位會使我的代碼停止作業。
用 netcat 替換客戶端或服務器表明在這兩種情況下，資料都按預期通過套接字發送
如果對此沒有解決方案，我愿意了解將資料攝取到 Spark 中的替代方法，這些方法也可以在這種情況下使用
uj5u.com熱心網友回復：
如檔案中所述，文本流中的記錄（行）由新行（）分隔\n。與print(),不同的是，sendall()它是一個面向位元組的函式，它不會自動添加新行。無論您發送多少標簽，Spark 都會繼續將所有內容作為一條記錄讀取，等待分隔符出現。當您改為發送推文文本時，它會起作用，因為某些推文確實包含換行符。
嘗試以下方法，看看它是否有效：
tcp_conn.sendall((tweet '\n').encode('utf-8'))
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/462597.html