6.4 具体实现代码

// 2.5 写入到 Phoenix(HBase)
import org.apache.phoenix.spark._
filteredStartupLogDStream.foreachRDD(rdd => {
    rdd.foreach(log => {
        println(log.logType)
    })
    // 参数1: 表名  参数2: 列名组成的 seq 参数 zkUrl: zookeeper 地址
    rdd.saveToPhoenix(
        "GMALL_DAU",
        Seq("MID", "UID", "APPID", "AREA", "OS", "CHANNEL", "LOGTYPE", "VERSION", "TS", "LOGDATE", "LOGHOUR"),
        zkUrl = Some("hadoop201,hadoop202,hadoop203:2181"))
})