Почему rdd всегда пуст во время загрузки данных Кафки в реальном времени в HBase через PySpark? - PullRequest
Новая
       51

Почему rdd всегда пуст во время загрузки данных Кафки в реальном времени в HBase через PySpark?

0 голосов
/ 04 марта 2019

Я пытаюсь выполнить в реальном времени ввод данных Kafka в HBase через PySpark в соответствии с этого руководства .Кажется, все работает нормально.Я запускаю кафку sudo /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties, затем запускаю продюсер /usr/local/kafka/bin/kafka-console-producer.sh --broker-list=myserver:9092 --topic test.Затем я запускаю исходный код, показанный ниже.Я отправляю сообщения производителю, но rdd.isEmpty(): всегда пусто.Так что я не достигаю линии с print("=some records=").Есть идеи что не так? 

appName = "Kafka_MapR-Streams_to_HBase"
config = SparkConf().setAppName(appName)

props = []
props.append(("spark.rememberDuration", "10"))
props.append(("spark.batchDuration", "10"))
props.append(("spark.eventLog.enabled", "true"))
props.append(("spark.streaming.timeout", "3000"))
props.append(("spark.ui.enabled", "true"))

config = config.setAll(props)

sc = SparkContext(conf=config)
ssc = StreamingContext(sc, int(config.get("spark.batchDuration")))


############################################################
# Main application execution function
############################################################

def runApplication(ssc, config):
  ssc.start()
  if config.get("spark.streaming.timeout") == '':
    ssc.awaitTermination()
  else:
    stopped = ssc.awaitTerminationOrTimeout(int(config.get("spark.streaming.timeout")))
  if not stopped :
    print("Stopping streaming context after timeout...")
    ssc.stop(True)
    print("Streaming context stopped.")


############################################################
# Output application details
############################################################

print
print( "APPNAME:" + config.get( "spark.app.name" ))
print( "APPID:" +   sc.applicationId)
print( "VERSION:" + sc.version)
print


############################################################
# Begin of Mapping Logic
############################################################

# HBase related details
hbase_table = 'clicks'
hconn = happybase.Connection('myserver')
ctable = hconn.table(hbase_table)
##

# MapR Streams / Kafka related details
topic = ["test"]
k_params = {"key.deserializer" : "org.apache.kafka.common.serialization.StringDeserializer" \
    ,"value.deserializer" : "org.apache.kafka.common.serialization.StringDeserializer" \
    #,"zookeeper.connect" : "maprdemo:5181"
    ,"metadata.broker.list" : "myserver:9092"
    ,"session.timeout.ms" : "45"
    ,"group.id" : "Kafka_MapR-Streams_to_HBase"}
##

def SaveToHBase(rdd):
    print("=====Pull from Stream=====")
    if not rdd.isEmpty():
        print("=some records=")
        for line in rdd.collect():
            ctable.put(('click' + line.serial_id), { \
            b'clickinfo:studentid': (line.studentid), \
            b'clickinfo:url': (line.url), \
            b'clickinfo:time': (line.time), \
            b'iteminfo:itemtype': (line.itemtype), \
            b'iteminfo:quantity': (line.quantity)})


kds = KafkaUtils.createDirectStream(ssc, topic, k_params, fromOffsets=None)

parsed = kds.filter(lambda x: x != None and len(x) > 0 )
parsed = parsed.map(lambda x: x[1])
parsed = parsed.map(lambda rec: rec.split(","))
parsed = parsed.filter(lambda x: x != None and len(x) == 6 )
parsed = parsed.map(lambda data:Row(serial_id=getValue(str,data[0]), \
        studentid=getValue(str,data[1]), \
        url=getValue(str,data[2]), \
        time=getValue(str,data[3]), \
        itemtype=getValue(str,data[4]), \
        quantity=getValue(str,data[5])))

parsed.foreachRDD(SaveToHBase)


############################################################
# Start application
############################################################

runApplication(ssc, config)

print
print("SUCCESS")
print
...