摘要：一直接訪問引入的相關包使用代替給指定配置與訪問本地文件一樣訪問文件實際測試中發現本地如能夠成功讀寫，但是集群模式下如讀寫失敗，原因未知。二通過訪問除了直接讀寫的數據，還可以通過來進行讀寫。

1.引入HDFS的相關jar包：

    
    org.apache.beam
    beam-sdks-java-io-hadoop-file-system
    2.1.0
    

2.使用HadoopFileSystemOptions代替PipelineOptions

public interface WordCountOptions extends HadoopFileSystemOptions {
    @Description("input file")
    @Default.String("hdfs://localhost:9000/tmp/words2")
    String getInputFile();
    void setInputFile(String in);

    @Description("output")
    @Default.String("hdfs://localhost:9000/tmp/hdfsWordCount")
    String getOutput();
    void setOutput(String out);
}

3.給Options指定HDFS配置

    Configuration conf=new Configuration();
    conf.set("fs.default.name", "hdfs://localhost:9000");
    HDFSOption options= PipelineOptionsFactory.fromArgs(args).withValidation()
            .as(HDFSOption.class);
    options.setHdfsConfiguration(ImmutableList.of(conf));
    

4.與訪問本地文件一樣訪問HDFS文件

    Pipeline p = Pipeline.create(options);
    Data = p.apply("Read from HDFS",
            TextIO.read().from(options.getInputFile()));
            

實際測試中發現本地runner（如Direct, Flink Local, Spark Local...）能夠成功讀寫HDFS，但是集群模式下（如Flink Cluster, Spark Cluster...）讀寫HDFS失敗，原因未知。

除了直接讀寫HDFS的數據，還可以通過HBase來進行讀寫。
1.添加相關jar包

    
    
        org.apache.beam
        beam-sdks-java-io-hbase
        ${beam.verson}
    

2.設置HBase連接信息

    Configuration conf = new Configuration();
    conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
    conf.setStrings("hbase.master.hostname", "localhost");
    conf.setStrings("hbase.regionserver.hostname", "localhost");

3.使用上述的conf讀HBase數據

    pipe
            //指定配置和表名
            .apply("Read from HBase",
                    HBaseIO.read().withConfiguration(conf).withTableId("test_tb"))
                       
            .apply(ParDo.of(new DoFn() {
                @ProcessElement
                public void processElement(ProcessContext c) {
                 //讀到的數據是HBase API中定義的Result格式，需要按照HBase官方說明進行剝取 
                    Result result = c.element();
                    String rowkey = Bytes.toString(result.getRow());
                    System.out.println("row key: ");

                    for(Cell cell : result.listCells())
                    {
                        System.out.println("qualifier:"+Bytes.toString(CellUtil.cloneQualifier(cell)));
                        System.out.println("value:"+Bytes.toString(CellUtil.cloneValue(cell)));
                    }

                    c.output(rowkey);
                }
            }));

4.寫入到HBase

    //寫入前需要將string數據封裝為Hbase數據格式mutation
    .apply(ParDo.of(new DoFn() {
        @ProcessElement
        public void processElement(ProcessContext context) {
            byte[] qual = Bytes.toBytes("qual");
            byte[] cf = Bytes.toBytes("cf");
            byte[] row = Bytes.toBytes("kafka");
            byte[] val = Bytes.toBytes(context.element());
            final Charset UTF_8 = Charset.forName("UTF-8");
            Mutation mutation = new Put(row).addColumn(cf, qual, val);
            context.output(mutation);
        }

    }))
    .apply("write to Hbase",
            HBaseIO.write()
                    .withConfiguration(conf)
                    .withTableId("test_tb"));

經測試，無論本地runner還是集群runner都能成功讀寫。
但是發現一個問題，使用mvn exec:java進行調試成功，而使用shade插件打包成jar運行卻一直報錯，說Mutation沒有指定coder，beam論壇上求助后得到的回復是maven-shade-plugin版本太舊，需要更新到3.0.0以上版本，但我改了3.0的版本之后還是一樣的錯誤。后來添加了ServicesResourceTransformer才解決。