title: storm hdfs connector ... trying to write data into hdfs using storm
body: <p>I am trying to write data into HDFS using "storm-hdfs connector 0.1.3". 
  The github URL: <a href="https://github.com/ptgoetz/storm-hdfs" rel="nofollow">https://github.com/ptgoetz/storm-hdfs</a>,I have added this dependency into my maven project. </p>
  
  <pre><code>&lt;dependency&gt;
        &lt;groupId&gt;com.github.ptgoetz&lt;/groupId&gt;
        &lt;artifactId&gt;storm-hdfs&lt;/artifactId&gt;
        &lt;version&gt;0.1.3-SNAPSHOT&lt;/version&gt;
        &lt;scope&gt;provided&lt;/scope&gt;
  &lt;/dependency&gt;
  </code></pre>
  
  <p>The sample topology to write the data into HDFS is provided in the storm-hdfs project itself. I just modified it to match to my file locations. The HdfsFileTopology is : </p>
  
  <pre><code>package my.company.app;
  
  import backtype.storm.Config;
  import backtype.storm.LocalCluster;
  import backtype.storm.StormSubmitter;
  import backtype.storm.spout.SpoutOutputCollector;
  import backtype.storm.task.OutputCollector;
  import backtype.storm.task.TopologyContext;
  import backtype.storm.topology.OutputFieldsDeclarer;
  import backtype.storm.topology.TopologyBuilder;
  import backtype.storm.topology.base.BaseRichBolt;
  import backtype.storm.topology.base.BaseRichSpout;
  import backtype.storm.tuple.Fields;
  import backtype.storm.tuple.Tuple;
  import backtype.storm.tuple.Values;
  import org.apache.storm.hdfs.bolt.HdfsBolt;
  import org.apache.storm.hdfs.bolt.AbstractHdfsBolt;
  import org.apache.storm.hdfs.bolt.SequenceFileBolt;
  import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;
  import org.apache.storm.hdfs.bolt.format.DelimitedRecordFormat;
  import org.apache.storm.hdfs.bolt.format.FileNameFormat;
  import org.apache.storm.hdfs.bolt.format.RecordFormat;
  import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
  import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy;
  import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy.Units;
  import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy;
  import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
  import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
  import org.apache.storm.hdfs.common.rotation.MoveFileAction;
  import org.yaml.snakeyaml.Yaml;
  
  import java.io.FileInputStream;
  import java.io.InputStream;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.UUID;
  import java.util.concurrent.ConcurrentHashMap;
  
  public class HdfsFileTopology {
  static final String SENTENCE_SPOUT_ID = "sentence-spout";
  static final String BOLT_ID = "my-bolt";
  static final String TOPOLOGY_NAME = "test-topology";
  
  public static void main(String[] args) throws Exception {
      Config config = new Config();
      config.setNumWorkers(1);
  
      SentenceSpout spout = new SentenceSpout();
  
      // sync the filesystem after every 1k tuples
      SyncPolicy syncPolicy = new CountSyncPolicy(1000);
  
      // rotate files when they reach 5MB
      FileRotationPolicy rotationPolicy = new TimedRotationPolicy(1.0f, TimedRotationPolicy.TimeUnit.MINUTES);
  
     FileNameFormat fileNameFormat = new DefaultFileNameFormat()
              .withPath("/users/storm/")
              .withExtension(".txt");
  
  
  
      // use "|" instead of "," for field delimiter
      RecordFormat format = new DelimitedRecordFormat()
              .withFieldDelimiter("|");
  
      Yaml yaml = new Yaml();
      InputStream in = new FileInputStream(args[1]);
      Map&lt;String, Object&gt; yamlConf = (Map&lt;String, Object&gt;) yaml.load(in);
      in.close();
      config.put("hdfs.config", yamlConf);
  
      HdfsBolt bolt = new HdfsBolt()
              .withConfigKey("hdfs.config")
              .withFsUrl(args[0])
              .withFileNameFormat(fileNameFormat)
              .withRecordFormat(format)
              .withRotationPolicy(rotationPolicy)
              .withSyncPolicy(syncPolicy)
              .addRotationAction(new MoveFileAction().toDestination("/dest2/"));
  
      TopologyBuilder builder = new TopologyBuilder();
  
      builder.setSpout(SENTENCE_SPOUT_ID, spout, 1);
      // SentenceSpout --&gt; MyBolt
      builder.setBolt(BOLT_ID, bolt, 4)
              .shuffleGrouping(SENTENCE_SPOUT_ID);
  
      if (args.length == 2) {
          LocalCluster cluster = new LocalCluster();
  
          cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
          waitForSeconds(120);
          cluster.killTopology(TOPOLOGY_NAME);
          cluster.shutdown();
          System.exit(0);
      } else if (args.length == 3) {
          StormSubmitter.submitTopology(args[0], config, builder.createTopology());
      } else{
          System.out.println("Usage: HdfsFileTopology [topology name] &lt;yaml config file&gt;");
      }
  }
  
  public static void waitForSeconds(int seconds) {
      try {
          Thread.sleep(seconds * 1000);
      } catch (InterruptedException e) {
      }
  }
  
  public static class SentenceSpout extends BaseRichSpout {
      private ConcurrentHashMap&lt;UUID, Values&gt; pending;
      private SpoutOutputCollector collector;
      private String[] sentences = {
              "my dog has fleas",
              "i like cold beverages",
              "the dog ate my homework",
              "don't have a cow man",
              "i don't think i like fleas"
      };
      private int index = 0;
      private int count = 0;
      private long total = 0L;
  
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("sentence", "timestamp"));
      }
  
      public void open(Map config, TopologyContext context,
                       SpoutOutputCollector collector) {
          this.collector = collector;
          this.pending = new ConcurrentHashMap&lt;UUID, Values&gt;();
      }
  
      public void nextTuple() {
          Values values = new Values(sentences[index], System.currentTimeMillis());
          UUID msgId = UUID.randomUUID();
          this.pending.put(msgId, values);
          this.collector.emit(values, msgId);
          index++;
          if (index &gt;= sentences.length) {
              index = 0;
          }
          count++;
          total++;
          if(count &gt; 20000){
              count = 0;
              System.out.println("Pending count: " + this.pending.size() + ", total: " + this.total);
          }
          Thread.yield();
      }
  
      public void ack(Object msgId) {
          this.pending.remove(msgId);
      }
  
      public void fail(Object msgId) {
          System.out.println("**** RESENDING FAILED TUPLE");
          this.collector.emit(this.pending.get(msgId), msgId);
      }
  }
  
  public static class MyBolt extends BaseRichBolt {
  
      private HashMap&lt;String, Long&gt; counts = null;
      private OutputCollector collector;
  
      public void prepare(Map config, TopologyContext context, OutputCollector collector) {
          this.counts = new HashMap&lt;String, Long&gt;();
          this.collector = collector;
      }
  
      public void execute(Tuple tuple) {
          collector.ack(tuple);
      }
  
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
          // this bolt does not emit anything
      }
  
      @Override
      public void cleanup() {
      }
  }
  }
  </code></pre>
  
  <p>I compile the project using maven (group id : my.company.app) and the build is successful but when I submit the jar file to storm it throws error.</p>
  
  <pre><code>Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/storm/hdfs/bolt/format/FileNameFormat
  Caused by: java.lang.ClassNotFoundException: org.apache.storm.hdfs.bolt.format.FileNameFormat
  </code></pre>
  
  <p>Even though I have included the class,why is it throwing an error that the class was not found? 
  And any help in how to write data to HDFS using storm will be appreciated. </p>
  
  <p>As requested the pom.xml </p>
  
  <pre><code>&lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
    &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
  
    &lt;groupId&gt;com.mycompany.app&lt;/groupId&gt;
    &lt;artifactId&gt;hdfs_example&lt;/artifactId&gt;
    &lt;version&gt;1.0-SNAPSHOT&lt;/version&gt;
    &lt;packaging&gt;jar&lt;/packaging&gt;
  
    &lt;name&gt;hdfs_example&lt;/name&gt;
    &lt;url&gt;http://maven.apache.org&lt;/url&gt;
  
    &lt;properties&gt;
      &lt;project.build.sourceEncoding&gt;UTF-8&lt;/project.build.sourceEncoding&gt;
    &lt;/properties&gt;
  
    &lt;repositories&gt;
          &lt;repository&gt;
            &lt;id&gt;Codehaus&lt;/id&gt;
            &lt;url&gt;http://repository.codehaus.org&lt;/url&gt;
          &lt;/repository&gt;
  
          &lt;repository&gt;
            &lt;id&gt;Codehaus.Snapshots&lt;/id&gt;
            &lt;url&gt;http://snapshots.repository.codehaus.org&lt;/url&gt;
            &lt;snapshots&gt;&lt;enabled&gt;true&lt;/enabled&gt;&lt;/snapshots&gt;
          &lt;/repository&gt;
  
          &lt;repository&gt;
              &lt;id&gt;github-releases&lt;/id&gt;
              &lt;url&gt;http://oss.sonatype.org/content/repositories/github-releases/&lt;/url&gt;
          &lt;/repository&gt;
  
          &lt;repository&gt;
              &lt;id&gt;clojars.org&lt;/id&gt;
              &lt;url&gt;http://clojars.org/repo&lt;/url&gt;
          &lt;/repository&gt;
      &lt;/repositories&gt;
  
    &lt;dependencies&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.testng&lt;/groupId&gt;
        &lt;artifactId&gt;testng&lt;/artifactId&gt;
        &lt;version&gt;6.8.5&lt;/version&gt;
        &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.mockito&lt;/groupId&gt;
        &lt;artifactId&gt;mockito-all&lt;/artifactId&gt;
        &lt;version&gt;1.9.0&lt;/version&gt;
        &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.easytesting&lt;/groupId&gt;
        &lt;artifactId&gt;fest-assert-core&lt;/artifactId&gt;
        &lt;version&gt;2.0M8&lt;/version&gt;
        &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;org.jmock&lt;/groupId&gt;
        &lt;artifactId&gt;jmock&lt;/artifactId&gt;
        &lt;version&gt;2.6.0&lt;/version&gt;
        &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
      &lt;groupId&gt;org.apache.storm&lt;/groupId&gt;
      &lt;artifactId&gt;storm-core&lt;/artifactId&gt;
      &lt;version&gt;0.9.2-incubating&lt;/version&gt;
          &lt;scope&gt;provided&lt;/scope&gt; 
  &lt;/dependency&gt;
  
      &lt;dependency&gt;
        &lt;groupId&gt;commons-collections&lt;/groupId&gt;
        &lt;artifactId&gt;commons-collections&lt;/artifactId&gt;
        &lt;version&gt;3.2.1&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;com.google.guava&lt;/groupId&gt;
        &lt;artifactId&gt;guava&lt;/artifactId&gt;
        &lt;version&gt;15.0&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;junit&lt;/groupId&gt;
        &lt;artifactId&gt;junit&lt;/artifactId&gt;
        &lt;version&gt;3.8.1&lt;/version&gt;
        &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
      &lt;groupId&gt;org.apache.hadoop&lt;/groupId&gt;
      &lt;artifactId&gt;hadoop-client&lt;/artifactId&gt;
      &lt;version&gt;2.2.0&lt;/version&gt;
      &lt;exclusions&gt;
          &lt;exclusion&gt;
              &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
              &lt;artifactId&gt;slf4j-log4j12&lt;/artifactId&gt;
          &lt;/exclusion&gt;
      &lt;/exclusions&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
        &lt;groupId&gt;com.github.ptgoetz&lt;/groupId&gt;
        &lt;artifactId&gt;storm-hdfs&lt;/artifactId&gt;
        &lt;version&gt;0.1.3-SNAPSHOT&lt;/version&gt;
            &lt;scope&gt;provided&lt;/scope&gt;
          &lt;/dependency&gt;
  
    &lt;/dependencies&gt;
  
    &lt;build&gt;
      &lt;sourceDirectory&gt;src/main/java&lt;/sourceDirectory&gt;
      &lt;testSourceDirectory&gt;test/main/java&lt;/testSourceDirectory&gt;
      &lt;resources&gt;
        &lt;resource&gt;
          &lt;directory&gt;${basedir}/multilang&lt;/directory&gt;
        &lt;/resource&gt;
      &lt;/resources&gt;
      &lt;plugins&gt;
      &lt;plugin&gt;
      &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
      &lt;artifactId&gt;maven-shade-plugin&lt;/artifactId&gt;
      &lt;version&gt;1.4&lt;/version&gt;
      &lt;configuration&gt;
          &lt;createDependencyReducedPom&gt;true&lt;/createDependencyReducedPom&gt;
          &lt;descriptorRefs&gt;
              &lt;descriptorRef&gt;jar-with-dependencies&lt;/descriptorRef&gt;
          &lt;/descriptorRefs&gt;
      &lt;/configuration&gt;
      &lt;executions&gt;
          &lt;execution&gt;
              &lt;phase&gt;package&lt;/phase&gt;
              &lt;goals&gt;
                  &lt;goal&gt;shade&lt;/goal&gt;
              &lt;/goals&gt;
              &lt;configuration&gt;
                  &lt;transformers&gt;
                      &lt;transformer
                              implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/&gt;
                      &lt;transformer
                              implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"&gt;
                          &lt;mainClass&gt;&lt;/mainClass&gt;
                      &lt;/transformer&gt;
                  &lt;/transformers&gt;
              &lt;/configuration&gt;
          &lt;/execution&gt;
      &lt;/executions&gt;
      &lt;/plugin&gt;
      &lt;!--  
      &lt;plugin&gt;
          &lt;groupId&gt;com.theoryinpractise&lt;/groupId&gt;
          &lt;artifactId&gt;clojure-maven-plugin&lt;/artifactId&gt;
          &lt;version&gt;1.3.12&lt;/version&gt;
          &lt;extensions&gt;true&lt;/extensions&gt;
          &lt;configuration&gt;
            &lt;sourceDirectories&gt;
              &lt;sourceDirectory&gt;src/clj&lt;/sourceDirectory&gt;
            &lt;/sourceDirectories&gt;
          &lt;/configuration&gt;
          &lt;executions&gt;
            &lt;execution&gt;
              &lt;id&gt;compile&lt;/id&gt;
              &lt;phase&gt;compile&lt;/phase&gt;
              &lt;goals&gt;
                &lt;goal&gt;compile&lt;/goal&gt;
              &lt;/goals&gt;
            &lt;/execution&gt;
            &lt;execution&gt;
              &lt;id&gt;test&lt;/id&gt;
              &lt;phase&gt;test&lt;/phase&gt;
              &lt;goals&gt;
                &lt;goal&gt;test&lt;/goal&gt;
              &lt;/goals&gt;
            &lt;/execution&gt;
          &lt;/executions&gt;
        &lt;/plugin&gt;
     --&gt;
        &lt;plugin&gt;
          &lt;groupId&gt;org.codehaus.mojo&lt;/groupId&gt;
          &lt;artifactId&gt;exec-maven-plugin&lt;/artifactId&gt;
          &lt;version&gt;1.2.1&lt;/version&gt;
          &lt;executions&gt;
            &lt;execution&gt;
              &lt;goals&gt;
                &lt;goal&gt;exec&lt;/goal&gt;
              &lt;/goals&gt;
            &lt;/execution&gt;
          &lt;/executions&gt;
          &lt;configuration&gt;
            &lt;executable&gt;java&lt;/executable&gt;
            &lt;includeProjectDependencies&gt;true&lt;/includeProjectDependencies&gt;
            &lt;includePluginDependencies&gt;false&lt;/includePluginDependencies&gt;
            &lt;classpathScope&gt;compile&lt;/classpathScope&gt;
            &lt;mainClass&gt;${storm.topology}&lt;/mainClass&gt;
          &lt;/configuration&gt;
        &lt;/plugin&gt;
  
        &lt;plugin&gt;
          &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
          &lt;artifactId&gt;maven-compiler-plugin&lt;/artifactId&gt;
          &lt;configuration&gt;
            &lt;source&gt;1.6&lt;/source&gt;
            &lt;target&gt;1.6&lt;/target&gt;
          &lt;/configuration&gt;
        &lt;/plugin&gt;
  
      &lt;/plugins&gt;
    &lt;/build&gt;  
  &lt;/project&gt;
  </code></pre>