Mysql 流增量写入 Hdfs（二） --Storm + hdfs 的流式处理
一. 概述 上一篇我们介绍了如何将数据从 mysql 抛到 kafka，这次我们就专注于利用 storm 将数据写入到 hdfs 的过程，由于 storm 写入 hdfs 的可定制东西有些多，我们先不从 kafka 读取，而先自己定义一个 Spout 数据充当数据源，下章再进行整合。这里默认你是拥有一定的 storm 知识的基础，起码知道 Spout 和 bolt 是什么。 写入 hdfs 可以有以下的定制策略： 自定义写入文件的名字 定义写入内容格式 满足给定条件后更改写入的文件 更改写入文件时触发的 Action 本篇会先说明如何用 storm 写入 HDFS，写入过程一些 API 的描述，以及最后给定一个例子： storm 每接收到 10 个 Tuple 后就会改变 hdfs 写入文件，新文件的名字就是第几次改变。 ps：storm 版本：1.1.1 。Hadoop 版本：2.7.4 。 接下来我们首先看看 Storm 如何写入 HDFS 。 二. Storm 写入 HDFS Storm 官方有提供了相应的 API 让我们可以使用。可以通过创建 HdfsBolt 以及定义相应的规则，即可写入 HDFS 。 首先通过 maven 配置依赖以及插件。 <properties> <storm.version>1.1.1</storm.version> </properties> <dependencies> <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>${storm.version}</version> <!--<scope>provided</scope>--> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>log4j-over-slf4j</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>commons-collections</groupId> <artifactId>commons-collections</artifactId> <version>3.2.1</version> </dependency> <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>15.0</version> </dependency> <!--hadoop模块--> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.7.4</version> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.7.4</version> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> </exclusions> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.storm/storm-hdfs --> <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-hdfs</artifactId> <version>1.1.1</version> <!--<scope>test</scope>--> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.5.1</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> <plugin> <groupId>org.codehaus.mojo</groupId> <artifactId>exec-maven-plugin</artifactId> <version>1.2.1</version> <executions> <execution> <goals> <goal>exec</goal> </goals> </execution> </executions> <configuration> <executable>java</executable> <includeProjectDependencies>true</includeProjectDependencies> <includePluginDependencies>false</includePluginDependencies> <classpathScope>compile</classpathScope> <mainClass>com.learningstorm.kafka.KafkaTopology</mainClass> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>1.7</version> <configuration> <createDependencyReducedPom>true</createDependencyReducedPom> </configuration> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass></mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build> 这里要提一下，如果要打包部署到集群上的话，打包的插件需要使用 maven-shade-plugin 这个插件，然后使用 maven Lifecycle 中的 package 打包。而不是用 Maven-assembly-plugin 插件进行打包。 因为使用 Maven-assembly-plugin 的时候，会将所有依赖的包unpack，然后在pack，这样就会出现，同样的文件被覆盖的情况。发布到集群上的时候就会报 No FileSystem for scheme: hdfs 的错 。 然后是使用 HdfsBolt 写入 Hdfs。这里来看看官方文档中的例子吧。 // 使用 "|" 来替代 ","，来进行字符分割 RecordFormat format = new DelimitedRecordFormat() .withFieldDelimiter("|"); // 每输入 1k 后将内容同步到 Hdfs 中 SyncPolicy syncPolicy = new CountSyncPolicy(1000); // 当文件大小达到 5MB ，转换写入文件，即写入到一个新的文件中 FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB); //当转换写入文件时，生成新文件的名字并使用 FileNameFormat fileNameFormat = new DefaultFileNameFormat() .withPath("/foo/"); HdfsBolt bolt = new HdfsBolt() .withFsUrl("hdfs://localhost:9000") .withFileNameFormat(fileNameFormat) .withRecordFormat(format) .withRotationPolicy(rotationPolicy) .withSyncPolicy(syncPolicy); //生成该 bolt topologyBuilder.setBolt("hdfsBolt", bolt, 5).globalGrouping("randomStrSpout"); 到这里就结束了。可以将 HdfsBolt 当作一个 Storm 中特殊一些的 bolt 即可。这个 bolt 的作用即使根据接收信息写入 Hdfs。 而在新建 HdfsBolt 中，Storm 为我们提供了相当强的灵活性，我们可以定义一些策略，比如当达成某个条件的时候转换写入文件，新写入文件的名字，写入时候的分隔符等等。 如果选择使用的话，Storm 有提供部分接口供我们使用，但如果我们觉得不够丰富也可以自定义相应的类。下面我们看看如何控制这些策略吧。 RecordFormat 这是一个接口，允许你自由定义接收到内容的格式。 public interface RecordFormat extends Serializable { byte[] format(Tuple tuple); } Storm 提供了 DelimitedRecordFormat ，使用方法在上面已经有了。这个类默认的分割符是逗号","，而你可以通过 withFieldDelimiter 方法改变分隔符。如果你的初始分隔符不是逗号的话，那么也可以重写写一个类实现 RecordFormat 接口即可。 FileNameFormat 同样是一个接口。 public interface FileNameFormat extends Serializable { void prepare(Map conf, TopologyContext topologyContext); String getName(long rotation, long timeStamp); String getPath(); } Storm 所提供的默认的是 org.apache.storm.hdfs.format.DefaultFileNameFormat 。默认人使用的转换文件名有点长，格式是这样的： {prefix}{componentId}-{taskId}-{rotationNum}-{timestamp}{extension} 例如： MyBolt-5-7-1390579837830.txt 默认情况下，前缀是空的，扩展标识是".txt"。 SyncPolicy 同步策略允许你将 buffered data 缓冲到 Hdfs 文件中（从而client可以读取数据），通过实现org.apache.storm.hdfs.sync.SyncPolicy 接口： public interface SyncPolicy extends Serializable { boolean mark(Tuple tuple, long offset); void reset(); } FileRotationPolicy 这个接口允许你控制什么情况下转换写入文件。 public interface FileRotationPolicy extends Serializable { boolean mark(Tuple tuple, long offset); void reset(); } Storm 有提供三个实现该接口的类： 最简单的就是不进行转换的org.apache.storm.hdfs.bolt.rotation.NoRotationPolicy ，就是什么也不干。 通过文件大小触发转换的 org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy。 通过时间条件来触发转换的 org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy。 如果有更加复杂的需求也可以自己定义。 RotationAction 这个主要是提供一个或多个 hook ，可加可不加。主要是在触发写入文件转换的时候会启动。 public interface RotationAction extends Serializable { void execute(FileSystem fileSystem, Path filePath) throws IOException; } 三.实现一个例子 了解了上面的情况后，我们会实现一个例子，根据写入记录的多少来控制写入转换（改变写入的文件），并且转换后文件的名字表示当前是第几次转换。 首先来看看 HdfsBolt 的内容： RecordFormat format = new DelimitedRecordFormat().withFieldDelimiter(" "); // sync the filesystem after every 1k tuples SyncPolicy syncPolicy = new CountSyncPolicy(1000); // FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(1.0f, FileSizeRotationPolicy.Units.KB); /** rotate file with Date,every month create a new file * format:yyyymm.txt */ FileRotationPolicy rotationPolicy = new CountStrRotationPolicy(); FileNameFormat fileNameFormat = new TimesFileNameFormat().withPath("/test/"); RotationAction action = new NewFileAction(); HdfsBolt bolt = new HdfsBolt() .withFsUrl("hdfs://127.0.0.1:9000") .withFileNameFormat(fileNameFormat) .withRecordFormat(format) .withRotationPolicy(rotationPolicy) .withSyncPolicy(syncPolicy) .addRotationAction(action); 然后分别来看各个策略的类。 FileRotationPolicy import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy; import org.apache.storm.tuple.Tuple; import java.text.SimpleDateFormat; import java.util.Date; /** * 计数以改变Hdfs写入文件的位置，当写入10次的时候，则更改写入文件，更改名字取决于 “TimesFileNameFormat” * 这个类是线程安全 */ public class CountStrRotationPolicy implements FileRotationPolicy { private SimpleDateFormat df = new SimpleDateFormat("yyyyMM"); private String date = null; private int count = 0; public CountStrRotationPolicy(){ this.date = df.format(new Date()); // this.date = df.format(new Date()); } /** * Called for every tuple the HdfsBolt executes. * * @param tuple The tuple executed. * @param offset current offset of file being written * @return true if a file rotation should be performed */ @Override public boolean mark(Tuple tuple, long offset) { count ++; if(count == 10) { System.out.print("num :" +count + " "); count = 0; return true; } else { return false; } } /** * Called after the HdfsBolt rotates a file. */ @Override public void reset() { } @Override public FileRotationPolicy copy() { return new CountStrRotationPolicy(); } } FileNameFormat import org.apache.storm.hdfs.bolt.format.FileNameFormat; import org.apache.storm.task.TopologyContext; import java.util.Map; /** * 决定重新写入文件时候的名字 * 这里会返回是第几次转换写入文件，将这个第几次做为文件名 */ public class TimesFileNameFormat implements FileNameFormat { //默认路径 private String path = "/storm"; //默认后缀 private String extension = ".txt"; private Long times = new Long(0); public TimesFileNameFormat withPath(String path){ this.path = path; return this; } @Override public void prepare(Map conf, TopologyContext topologyContext) { } @Override public String getName(long rotation, long timeStamp) { times ++ ; //返回文件名，文件名为更换写入文件次数 return times.toString() + this.extension; } public String getPath(){ return this.path; } } RotationAction import org.apache.hadoop.fs.FileContext; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.storm.hdfs.common.rotation.RotationAction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.net.URI; /** 当转换写入文件时候调用的 hook ，这里仅写入日志。 */ public class NewFileAction implements RotationAction { private static final Logger LOG = LoggerFactory.getLogger(NewFileAction.class); @Override public void execute(FileSystem fileSystem, Path filePath) throws IOException { LOG.info("Hdfs change the written file！！"); return; } } OK，这样就大功告成了。通过上面的代码，每接收到 10 个 Tuple 后就会转换写入文件，新文件的名字就是第几次转换。 完整代码包括一个随机生成字符串的 Spout ，可以到我的 github 上查看。 StormHdfsDemo:https://github.com/shezhiming/StormHdfsDemo 更多干货，欢迎关注公众号，哈尔的数据城堡。
