如何使用storm将数据持久化到HDFS答案

【问题标题】：How to persist data into HDFS using storm如何使用storm将数据持久化到HDFS
【发布时间】：2015-02-17 10:26:51
【问题描述】：

我有一个简单的螺栓，它从 kafka spout 读取数据，然后应该将数据写入 HDFS 目录。问题是在集群停止之前，bolt 不会写入。我如何确保当螺栓从 kafka spout 读取一个元组时，它会立即将其写入 HDFS，或者至少写入每个“n”个条目。（我使用的是 CDH 4.4、Hadoop 2.0）

螺栓的java：

public class PrinterBolt10 extends BaseRichBolt{  
    private OutputCollector collector;
    private String values;
    Configuration configuration = null;
    FileSystem hdfs = null;
    FSDataOutputStream outputStream=null;
    BufferedWriter br = null; 
    List<String> valList;
    String machineValue;
    int upTime;
    int downTime;
    int idleTime; 

    public void prepare(Map config, TopologyContext context,OutputCollector collector) {
        upTime=0;
        downTime=0;
        idleTime=0;
        this.collector = collector;
        String timeStamp = new SimpleDateFormat("yyyyMMdd_HHmmss").format(Calendar.getInstance().getTime());
        try{
            configuration = new Configuration();
            configuration.set("fs.defaultFS", "hdfs://localhost.localdomain:8020");
            hdfs =FileSystem.get(configuration);
            outputStream = hdfs.create(new Path("/tmp/storm/StormHdfs/machine10_"+timeStamp+".txt"));
            br = new BufferedWriter( new OutputStreamWriter( outputStream , "UTF-8" ) );
            br.flush(); 
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public void execute(Tuple tuple) {  
        values = tuple.toString();
        int start = values.indexOf('[');
        int end = values.indexOf(']'); 
        machineValue=values.substring(start+1,end); 
        String machine=machineValue.substring(0,machineValue.indexOf(','));
        String code = machineValue.substring(machineValue.indexOf(',')+1);
        int codeInt = Integer.parseInt(code);
        if(codeInt==0) idleTime+=30;
        elseif(codeInt==1) upTime+=30;
        else downTime+=30; 
        String finalMessage = machine + " "+ "upTime(s) :" + upTime+" "+ "idleTime(s): "+idleTime+" "+"downTime: "+downTime;  
        try {
            br.write(finalMessage);  // *This is the writing part into HDFS*
            br.write('\n'); 
            br.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // this bolt does not emit anything
    }

    public void cleanup() {}
}

【问题讨论】：

虽然这不是您问题的直接答案，但您也可以考虑将 Storm 输出写回 Kafka（您已经从 Kafka 读取，因此基础设施存在），然后使用诸如作为 LinkedIn Camus 负责（批量）将数据从 Kafka 加载到 HDFS。这种方法也可能更安全，因为从 Storm 直接写入 HDFS 可能会导致重复数据，请参阅blog.thedatateam.in/2015/02/guaranteeing-exactly-once-load.html。

标签： java hadoop hdfs apache-storm

【解决方案1】：

编辑：完全改变了我的答案。

您需要使用HdfsBolt 而不是依靠自己写文件。使用HdfsBolt 消除了计算何时刷新到文件、打开缓冲区流等的所有复杂性。请参阅http://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.1.3/bk_user-guide/content/ch_storm-using-hdfs-connector.html，但您感兴趣的位是：

// Use pipe as record boundary
RecordFormat format = new DelimitedRecordFormat().withFieldDelimiter("|");

//Synchronize data buffer with the filesystem every 1000 tuples
SyncPolicy syncPolicy = new CountSyncPolicy(1000);

// Rotate data files when they reach five MB
FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);

// Use default, Storm-generated file names
FileNameFormat fileNameFormat = new DefaultFileNameFormat().withPath("/foo");

// Instantiate the HdfsBolt
HdfsBolt bolt = new HdfsBolt()
     .withFsURL("hdfs://localhost:54310")
     .withFileNameFormat(fileNameFormat)
     .withRecordFormat(format)
     .withRotationPolicy(rotationPolicy)
     .withSyncPolicy(syncPolicy);

然后只需将您当前螺栓中的数据传递到此螺栓。

【讨论】：

是的，目的是我强制它把输出写到想要的目的地。
我又看了一遍，意识到我错了。结果我改变了答案。
我见过 hdfs-bolt ，但它导入了一个特定的类“org.apache.hadoop.hdfs.client.HdfsDataOutputStream.SyncFlag”，我似乎在 CDH 4.4 hadoop-hdfs 中不可用
@benwatsondata 我已经看过您的回答中给出的示例，但我很好奇它的实际用途。我是否需要单独的螺栓扩展 HhdfsBolt 类，然后进入 execute() 方法？这段代码是有道理的，我只是不知道它的去向。

【解决方案2】：

您应该使用 HdfsBolt 将数据插入 HDFS。使用作者描述的配置。与其将 SyncPolicy 计数设置为 1000，不如将其设置为某个最小数量（例如 10-20）以用于测试目的。因为该数字表示在 spout 发出多少个元组之后，这些元组应该写入 HDFS。例如，如果您配置了

SyncPolicy syncPolicy = new CountSyncPolicy(10);

那么您将能够在 10 条消息后看到您插入到 Kafka 的数据。

【讨论】：