如何将HDFS上的文件读入到Hbase，用java。调用insertrow方法

要将HDFS上的文件读入到HBase，并使用Insert操作将数据插入到HBase表中，可以使用Java编写MapReduce程序实现，以下是实现步骤：

1. 首先需要创建一个HBase表，可使用HBase Shell或Java API创建；
2. 编写MapReduce程序，其中Map阶段读取HDFS上的文件，将数据转换成Put对象，然后将Put对象写入到HBase表中；
3. 在MapReduce程序中设置HBase表名、列族名、列名等参数；
4. 在程序运行前，需要将HBase相关的jar包和配置文件加入到classpath中；
5. 最后提交MapReduce任务运行即可。

以下是示例代码：

Configuration conf = HBaseConfiguration.create();
Job job = Job.getInstance(conf, "HDFS to HBase");

job.setJarByClass(HdfsToHBase.class);
job.setMapperClass(HdfsToHBaseMapper.class);

FileInputFormat.addInputPath(job, new Path("hdfs://localhost:9000/input"));

TableMapReduceUtil.initTableReducerJob("table_name", null, job);
job.setOutputFormatClass(TableOutputFormat.class);

job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Put.class);

System.exit(job.waitForCompletion(true) ? 0 : 1);

其中，HdfsToHBaseMapper类的代码如下：

public class HdfsToHBaseMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {

    private final static byte[] CF = "cf".getBytes();
    private final static byte[] COLUMN = "column".getBytes();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split(",");
        String rowKey = fields[0];
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(CF, COLUMN, Bytes.toBytes(fields[1]));
        context.write(new ImmutableBytesWritable(Bytes.toBytes(rowKey)), put);
    }
}

在以上代码中，CF和COLUMN分别表示HBase表的列族名和列名，可以根据实际情况修改。同时，也要根据实际情况修改HDFS文件路径、HBase表名等参数。

如果需要使用Insert操作将数据插入到HBase表中，可以在HdfsToHBaseReducer类中实现，在reduce方法中逐条处理Put对象，使用Table对象的put方法将数据插入到HBase表中，代码如下：

public class HdfsToHBaseReducer extends TableReducer<ImmutableBytesWritable, Put, ImmutableBytesWritable> {

    private Table table;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        Configuration conf = HBaseConfiguration.create();
        table = new HTable(conf, "table_name");
    }

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
        for (Put put : values) {
            table.put(put);
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        table.close();
    }
}

在以上代码中，setup方法中创建Table对象，cleanup方法中关闭Table对象，遍历每个Put对象使用Table对象的put方法将数据插入到HBase表中。

需要注意的是，使用Insert操作将数据插入到HBase表中会有一定的性能损失，如果需要快速插入大量数据，建议使用Bulk Load操作。