如何将xml文件从hdfs加载到hbase表中

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

public class XmlToHBaseLoader {
    private static final String HDFS_URI = "hdfs://localhost:9000";
    private static final String HBASE_TABLE_NAME = "xml_data";
    private static final String HBASE_COLUMN_FAMILY = "data";

    public static void main(String[] args) {
        try {
            // 创建HDFS文件系统对象
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", HDFS_URI);
            FileSystem fs = FileSystem.get(conf);

            // 读取XML文件
            Path xmlPath = new Path("/path/to/xml/file.xml");
            DocumentBuilderFactory dbFactory = DocumentBuilderFactory.newInstance();
            DocumentBuilder dBuilder = dbFactory.newDocumentBuilder();
            Document doc = dBuilder.parse(fs.open(xmlPath));
            doc.getDocumentElement().normalize();

            // 创建HBase连接和表对象
            Connection connection = ConnectionFactory.createConnection();
            Table table = connection.getTable(TableName.valueOf(HBASE_TABLE_NAME));

            // 遍历XML文件中的节点
            NodeList nodeList = doc.getElementsByTagName("record");
            for (int i = 0; i < nodeList.getLength(); i++) {
                Node node = nodeList.item(i);
                if (node.getNodeType() == Node.ELEMENT_NODE) {
                    Element element = (Element) node;

                    // 解析XML节点中的数据
                    String id = element.getAttribute("id");
                    String name = element.getElementsByTagName("name").item(0).getTextContent();
                    String value = element.getElementsByTagName("value").item(0).getTextContent();

                    // 构造HBase表中的行键和列值
                    Put put = new Put(Bytes.toBytes(id));
                    put.addColumn(Bytes.toBytes(HBASE_COLUMN_FAMILY), Bytes.toBytes("name"), Bytes.toBytes(name));
                    put.addColumn(Bytes.toBytes(HBASE_COLUMN_FAMILY), Bytes.toBytes("value"), Bytes.toBytes(value));

                    // 写入数据到HBase表
                    table.put(put);
                }
            }

            // 关闭连接和表对象
            table.close();
            connection.close();

            System.out.println("XML data loaded into HBase successfully.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}