spark從hbase讀取數據以及zookeeper坑

1、遇到錯誤，認真查看日志，這才是解決問題的王道啊！

不過很多時候，尤其是開始摸索的時候，一下要接觸的東西太多了，學東西也太趕了，加上boss不停的催進度，結果欲速則不達，最近接觸大數據，由於平台是別人搭建的，搭沒搭好不知道，也不清楚細節，出了問題也不知道是自己這邊的還是平台的問題。有的時候就是小問題忽略了，結果花了好多時間又才重新發現。

 

提交job：

./spark-submit --class myapp.KMeansWeather --master yarn --deploy-mode cluster ./hbase_handles_cc.jar  (集群)

or

./spark-submit --class myapp.KMeansWeather --master local[1] ./hbase_handles_cc.jar (本地)

 

提交本地過后老是出現問題，就是程序不斷的提示錯誤，又不會停止，zookeeper客戶端連接一直報無法定位登錄配置信息。

INFO org.apache.zookeeper.ClientCnxn: Opening socket connection to server localhost/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)

還說是未知錯誤，我就更不知道錯誤在哪了，后來折騰了半天，第二天才看了下zookeeper的日志，zookeeper.out，如下：

2016-08-25 10:13:00,150 [myid:1] - INFO [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@192] - Accepted socket connection from /10.3.9.231:33070
2016-08-25 10:13:00,150 [myid:1] - WARN [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@362] - Exception causing close of session 0x0 due to java.io.IOException: ZooKeeperServer not running

然后我用./bin/zkServer.sh stop 與./bin/zkServer.sh start重新啟動，然后查看狀態，./bin/zkServer.sh status，發現

[root@hadoop bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /root/traffic-platform/zookeeper-3.4.8/bin/../conf/zoo.cfg
Error contacting service. It is probably not running.

於是向安裝環境那個人問了下，然后他最后說是集群里node3網線掉了…………

目前我們測試集群用了4台機子，有一台后來別人在用，所以配置的4台結果只有3台，分別是master,node2,node3，結果這次node3網線掉了，所以只有兩個節點了，因為zookeeper在機子少於集群的一半就無法選舉出leader來了吧。

把網線接上后，就可以了，再次查看每台機子的狀態。如下：一個是leader，其它的為follower

[root@hadoop bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /root/traffic-platform/zookeeper-3.4.8/bin/../conf/zoo.cfg
Mode: leader

其它：

查網上信息：說客戶端程序的zookeeper版本不同也會導致連接不上，一直報錯。

另外：看自己的hbase-site.xml的配置文件路徑是否加入到了編譯的path里面，當對Hbase進行操作時，它會指定相應的屬性包括集群ip,以及端口等

 

注意看紅色部分有那個黃色塊的表示加入到了path中了。

2、spark讀取Hbase

初學這個，在網上查了下資料，官網的資料也是難得翻所認沒翻到啊。

於是查看網上的信息來寫，與結合mllib中的JavaSparkPI算法

    public static void getValueFromHB() throws IOException {
        final Logger logger = LoggerFactory.getLogger(KMeansWeather.class);
        String f="BaseInfo";
        String table="NewCityWeather";
        Scan scan=new Scan();
        scan.addFamily(Bytes.toBytes(f));
        //scan.addColumn(f,)

        Configuration conf= HBaseConfiguration.create();//讀取hbase-site.xml等配置
        //conf.set("hbase.zookeeper.quorum","10.3.9.135,10.3.9.231,10.3.9.232");//這些在hbase-site.xml中是有的
        //conf.set("hbase.zookeeper.property.clientPort","2222");
        conf.set(TableInputFormat.INPUT_TABLE,table);//設置查詢的表
        conf.set(TableInputFormat.SCAN, Base64.encodeBytes(ProtobufUtil.toScan(scan).toByteArray()));//設置掃描的列
        //SparkConf confsp=new SparkConf().setAppName("SparkHBaseTest").setMaster("yarn-client");
        //SparkConf confsp=new SparkConf().setAppName("SparkHBaseTest").setMaster("spark://10.3.9.135:7077");

　　　　 //設置應用名稱，就是在spark web端顯示的應用名稱，當然還可以設置其它的，在提交的時候可以指定，所以不用set上面兩行吧

 　　　　SparkConf confsp=new SparkConf().setAppName("SparkHBaseTest");
　　　　 //創建spark操作環境對象
        JavaSparkContext sc = new JavaSparkContext(confsp);
//        JavaSparkContext sc = new JavaSparkContext("yarn-client", "hbaseTest",
//                System.getenv("SPARK_HOME"), System.getenv("JARS"));
        //sc.addJar("D:\\jiuzhouwork\\other\\sparklibex\\spark-examples-1.6.1-hadoop2.7.1.jar");
　　　　　　
　　　　　//從數據庫中獲取查詢內容生成RDD
        JavaPairRDD<ImmutableBytesWritable,Result> myRDD=sc.newAPIHadoopRDD(conf,TableInputFormat.class,ImmutableBytesWritable.class,Result.class);

　　　　　//遍歷數據 collect foreach
　　　　　List<Tuple2<ImmutableBytesWritable, Result>> output=myRDD.collect();
　　　　　for (Tuple2 tuple: output ) {
   　　　　　　 System.out.println(tuple._1+"："+tuple._2);
　　　　　}

　　　　 System.out.println("sss:"+myRDD.count());
        logger.info("lwwwww:"+myRDD.count());//輸出RDD數據條數
        //System.out.println("sss:");
        //logger.info("lwwwww:");


        JavaRDD rdd=JavaRDD.fromRDD(JavaPairRDD.toRDD(myRDD),myRDD.classTag());
//        JavaRDD<Vector> points = myRDD.map(new ParsePoint());
//
//        KMeansModel model = KMeans.train(points.rdd(), k, iterations, runs, KMeans.K_MEANS_PARALLEL());
//
//        System.out.println("Cluster centers:");
//        for (Vector center : model.clusterCenters()) {
//            System.out.println(" " + center);
//        }
//        double cost = model.computeCost(points.rdd());
//        System.out.println("Cost: " + cost);
//
//        sc.stop();


    }

 注意

SparkConf confsp=new SparkConf().setAppName("SparkHBaseTest")
        .setMaster("local")//以本地的形式運行
　　　　 .setMaster("spark://10.3.9.135:7077");//以standalone的方式運行，就是spark的集群調度方式

　　　　 .setMaster("yarn-client");//以yarn的集群方式運行，yarn

        .setJars(new String[]{"D:\\jiuzhouwork\\workspace\\hbase_handles\\out\\artifacts\\hbase_handles_jar\\hbase_handles.jar"});//設置提交的jar


遍歷數據：collect foreach 結果

問題： java.io.NotSerializableException: org.apache.hadoop.hbase.io.ImmutableBytes

由於沒有設置序列化的類。
在conf/spark-defaults.conf文件中設置

如果沒有spark-defaults.conf,則復制一份spark-defaults.conf.template更名為spark-defaults.conf

問題： 從HBASE中讀出數據信息但沒有值？

解決：通過Map,MapToPair,FlatMapToPair然后進行鍵值處理與映射，形成新的RDD,通過處理ImmutableBytesWritable.getValue(family,qualifier)來獲得信息。以前在Map中的操作對JavaPairRDD現在只能放在MapToPair里面了，所以如果用Map有錯，試着用一下MapToPair