更新时间:2022-09-08 GMT+08:00
分享

Storm-HDFS开发指引

操作场景

本章节只适用于Storm和HDFS交互的场景。本章中描述的jar包的具体版本信息请以实际情况为准。

应用开发操作步骤

  1. 确认Storm和HDFS组件已经安装,并正常运行。
  2. 将storm-examples导入到IntelliJ IDEA开发环境,请参见环境准备
  3. 下载并安装HDFS客户端。
  4. 获取相关配置文件。获取方法如下:

    在安装好的HDFS客户端目录下找到目录“/opt/clientHDFS/HDFS/hadoop/etc/hadoop”,在该目录下获取到配置文件“core-site.xml”和“hdfs-site.xml”。

  5. 获取相关jar包。获取方法如下:

    • 在安装好的HDFS客户端目录下找到目录“HDFS/hadoop/share/hadoop/common/lib”,获取如下jar包:
      • commons-cli-<version>.jar
      • commons-io-<version>.jar
      • commons-lang-<version>.jar
      • commons-lang3-<version>.jar
      • commons-collections-<version>.jar
      • commons-configuration2-<version>.jar
      • commons-logging-<version>.jar
      • guava-<version>.jar
      • hadoop-*.jar
      • protobuf-java-<version>.jar
      • jackson-databind-<version>.jar
      • jackson-core-<version>.jar
      • jackson-annotations-<version>.jar
      • re2j-<version>.jar
      • jaeger-core-<version>.jar
      • opentracing-api-<version>.jar
      • opentracing-noop-<version>.jar
      • opentracing-tracerresolver-<version>.jar
      • opentracing-util-<version>.jar
    • 在安装好的HDFS客户端目录下找到目录“HDFS/hadoop/share/hadoop/common”,获取该目录下的hadoop-*.jar。
    • 在安装好的HDFS客户端目录下找到目录“HDFS/hadoop/share/hadoop/client”,获取该目录下的hadoop-*.jar。
    • 在安装好的HDFS客户端目录下找到目录“HDFS/hadoop/share/hadoop/hdfs”,获取该目录下的hadoop-hdfs-*.jar。
    • 在样例工程“/src/storm-examples/storm-examples/lib”中获取如下jar包:
      • storm-hdfs-<version>.jar
      • storm-autocreds-<version>.jar

IntelliJ IDEA代码样例

创建Topology。

  public static void main(String[] args) throws Exception   
    {  
      TopologyBuilder builder = new TopologyBuilder();  

      // 分隔符格式,当前采用“|”代替默认的“,”对tuple中的field进行分隔  
      // HdfsBolt必选参数  
      RecordFormat format = new DelimitedRecordFormat()  
              .withFieldDelimiter("|");  

      // 同步策略,每1000个tuple对文件系统进行一次同步  
      // HdfsBolt必选参数  
      SyncPolicy syncPolicy = new CountSyncPolicy(1000);  

      // 文件大小循环策略,当文件大小到达5M时,从头开始写  
      // HdfsBolt必选参数  
      FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);  

      // 写入HDFS的目的文件  
      // HdfsBolt必选参数  
      FileNameFormat fileNameFormat = new DefaultFileNameFormat()  
              .withPath("/user/foo/");  


      //创建HdfsBolt  
      HdfsBolt bolt = new HdfsBolt()  
              .withFsUrl(DEFAULT_FS_URL)
              .withFileNameFormat(fileNameFormat)  
              .withRecordFormat(format)  
              .withRotationPolicy(rotationPolicy)  
              .withSyncPolicy(syncPolicy);  

      //Spout生成随机语句  
      builder.setSpout("spout", new RandomSentenceSpout(), 1);   
      builder.setBolt("split", new SplitSentence(), 1).shuffleGrouping("spout");  
      builder.setBolt("count", bolt, 1).fieldsGrouping("split", new Fields("word"));  

      //增加Kerberos认证所需的plugin到列表中,安全模式必选  
      setSecurityConf(conf,AuthenticationType.KEYTAB); 

      Config conf = new Config();  
      //将客户端配置的plugin列表写入config指定项中,安全模式必配  
      conf.put(Config.TOPOLOGY_AUTO_CREDENTIALS, auto_tgts);  

      if(args.length >= 2)  
      {  
          //用户更改了默认的keytab文件名,这里需要将新的keytab文件名通过参数传入  
          conf.put(Config.STORM_CLIENT_KEYTAB_FILE, args[1]);  
      }  

      //命令行提交拓扑  
      StormSubmitter.submitTopology(args[0], conf, builder.createTopology());  

    }

Storm不支持将HDFS的目的文件路径设置为HDFS的SM4加密分区。

部署运行及结果查看

  1. 导出本地jar包,请参见打包IntelliJ IDEA代码
  2. 1导出的本地Jar包,4中获取的配置文件和5中获取的jar包合并统一打出完整的业务jar包,请参见打包业务
  3. 执行命令提交拓扑。

    storm jar /opt/jartarget/source.jar com.huawei.storm.example.hdfs.SimpleHDFSTopology hdfs-test

  4. 拓扑提交成功后请登录HDFS集群查看。
分享:

    相关文档

    相关产品