SaveAsHadoopDataset 永远不会关闭与 zookeeper 的连接答案

【问题标题】：SaveAsHadoopDataset never closes connection To zookeeperSaveAsHadoopDataset 永远不会关闭与 zookeeper 的连接
【发布时间】：2016-09-22 22:35:39
【问题描述】：

我正在使用下面的代码写入 hbase

    jsonDStream.foreachRDD(new Function<JavaRDD<String>, Void>() {

        @Override
        public Void call(JavaRDD<String> rdd) throws Exception {

            DataFrame jsonFrame = sqlContext.jsonRDD(rdd);
            DataFrame selecteFieldFrame = jsonFrame.select("id_str","created_at","text");

            Configuration config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", "d-9543");
            config.set("zookeeper.znode.parent","/hbase-unsecure");
            config.set("hbase.zookeeper.property.clientPort", "2181");
            final JobConf jobConfig=new JobConf(config,SveAsHadoopDataSetExample.class);

            jobConfig.setOutputFormat(TableOutputFormat.class);
            jobConfig.set(TableOutputFormat.OUTPUT_TABLE,"tableName");
             selecteFieldFrame.javaRDD().mapToPair(new PairFunction<Row, ImmutableBytesWritable, Put>() {

                @Override
                public Tuple2<ImmutableBytesWritable, Put> call(Row row) throws Exception {
                    // TODO Auto-generated method stub
                    return convertToPut(row);
                }
            }).saveAsHadoopDataset(jobConfig);


            return null;
        }
    });

但是当我在 zookeeper 中看到 zkDump 时，连接不断增加

任何建议/指针都会有很大帮助！

【问题讨论】：

标签： hadoop apache-spark hbase apache-spark-sql spark-streaming

【解决方案1】：

我也有同样的问题，是hbase的bug，我修复了：

将 org.apache.hadoop.hbase.mapred.TableOutputFormat 更改为 org.apache.hadoop.hbase.mapreduce.TableOutputFormat，并使用 org.apache.hadoop.mapreduce.Job，而不是 org.apache.hadoop.mapred.JobConf

这是一个示例：

import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat

val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.quorum", zk_hosts)
conf.set("hbase.zookeeper.property.clientPort", zk_port)

conf.set(TableOutputFormat.OUTPUT_TABLE, "TABLE_NAME")
val job = Job.getInstance(conf)
job.setOutputFormatClass(classOf[TableOutputFormat[String]])

formatedLines.map{
  case (a,b, c) => {
    val row = Bytes.toBytes(a)

    val put = new Put(row)
    put.setDurability(Durability.SKIP_WAL)

    put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("node"), Bytes.toBytes(b))
    put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("topic"), Bytes.toBytes(c))

    (new ImmutableBytesWritable(row), put)
  }
}.saveAsNewAPIHadoopDataset(job.getConfiguration)

这可能对你有帮助！

https://github.com/hortonworks-spark/shc/pull/20/commits/2074067c42c5a454fa4cdeec18c462b5367f23b9

【讨论】：

虽然这在理论上可以回答这个问题，it would be preferable 在此处包含答案的基本部分，并提供链接以供参考。
谢谢你！仍然大多数手册都有存在错误的旧方法。我期待有stg。工作关闭（）。我的问题完全一样。