我尝试通过pyspark将kafka数据摄取到hbase中。目前，我尝试使用happybase将kafka的数据插入hbase，但是速度非常慢。我认为批量加载也使用happybase不会显著提高性能。下面是当前的源代码。我需要达到最高可能的表现。你有什么想法吗？也许 吧 saveAsNewAPIHadoopDataset 或者其他解决方案？

def SaveToHBase(rdd):
    print("=====Pull from Stream=====")
    if not rdd.isEmpty():

        print(len(rdd.collect()))
        print(datetime.now())
        for line in rdd.collect():
            ctable.put((line.log_id), { \
            b'log:content': (line.log)})

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) == 2 )
parsed = parsed.map(lambda data:Row(log_id=getValue(str,data[0]), \
        log=getValue(str,data[1])))

parsed.foreachRDD(SaveToHBase)