火花插入到 Hbase答案

【问题标题】：Spark insert to Hbase火花插入到 Hbase
【发布时间】：2016-07-31 13:12:59
【问题描述】：

我有一个 pojo 类的 emp，如下所示：我能够读取流数据并且我想将数据插入 Hbase

@JsonInclude(Include.NON_NULL)
public class empData implements Serializable {

    private String id;
    private String name;

    @Override
    public String toString() {
        return "id=" + id + ", name="+ name ;
    }
    public String id() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }

}

以下是火花代码：

empRecords.foreachRDD(new Function<JavaRDD<empData>, Void>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Void call(JavaRDD<empData> empDataEvent)throws Exception {       

                Configuration conf = HBaseConfiguration.create();
                Configuration config = null;
                config = HBaseConfiguration.create();
                config.set("hbase.zookeeper.quorum", "**********);
                HBaseAdmin.checkHBaseAvailable(config);
                config.set(TableInputFormat.INPUT_TABLE, "tableName");
                Job newAPIJobConfiguration1 = Job.getInstance(config);
                newAPIJobConfiguration1.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "empHbase");
                newAPIJobConfiguration1.setOutputFormatClass(org.apache.hadoop.hbase.mapreduce.TableOutputFormat.class);        
                JavaPairRDD<ImmutableBytesWritable, Put> inesrts = empData.mapToPair(new PairFunction<Row, ImmutableBytesWritable, Put>() {

                            public Tuple2<ImmutableBytesWritable, Put> call(Row row) throws Exception

                            {
                                Put put = new Put(Bytes.toBytes(row.getString(0)));
                                put.add(Bytes.toBytes("empA"),Bytes.toBytes("id"),Bytes.toBytes(row.getString(1)));
                                put.add(Bytes.toBytes("empA"),Bytes.toBytes("name"),Bytes.toBytes(row.getString(2)));
                                return new Tuple2<ImmutableBytesWritable, Put>(new ImmutableBytesWritable(), put);
                            }
                                });

                            inserts.saveAsNewAPIHadoopDataset(newAPIJobConfiguration1.getConfiguration());
                                        }
        });
        jssc.start();
        jssc.awaitTermination();
    }

代码中的问题是这一步：

JavaPairRDD<ImmutableBytesWritable, Put> inesrts =empDataEvent.mapToPair(new PairFunction<Row, ImmutableBytesWritable, Put>()

如何使用empDataEvent以及如何插入.. 如何插入为 mapToPair empDataEvent 类对象，以便我可以插入 Hbase。任何帮助表示赞赏..

【问题讨论】：

标签： hadoop apache-spark hbase spark-streaming

【解决方案1】：

阿曼，

在您的代码中，您引用了“Row”，您能否详细说明它的来源？因为没有参考。

查看下面的更新代码，使用类名“empData”而不是“Row”对象。

JavaPairRDD<ImmutableBytesWritable, Put> inesrts = empData.mapToPair(new PairFunction<empData, ImmutableBytesWritable, Put>() {

                        public Tuple2<ImmutableBytesWritable, Put> call(empData row) throws Exception
                        {
                            Put put = new Put(Bytes.toBytes(row.id));
                            put.add(Bytes.toBytes("empA"),Bytes.toBytes("id"),Bytes.toBytes(row.id));
                            put.add(Bytes.toBytes("empA"),Bytes.toBytes("name"),Bytes.toBytes(row.getName));
                            return new Tuple2<ImmutableBytesWritable, Put>(new ImmutableBytesWritable(), put);
                        }
                            });

【讨论】：