您现在的位置是:首页 > 文章详情

Mysql 流增量写入 Hdfs(二) --Storm + hdfs 的流式处理

日期:2018-12-11点击:279

一. 概述

上一篇我们介绍了如何将数据从 mysql 抛到 kafka,这次我们就专注于利用 storm 将数据写入到 hdfs 的过程,由于 storm 写入 hdfs 的可定制东西有些多,我们先不从 kafka 读取,而先自己定义一个 Spout 数据充当数据源,下章再进行整合。这里默认你是拥有一定的 storm 知识的基础,起码知道 Spout 和 bolt 是什么。

写入 hdfs 可以有以下的定制策略:

  1. 自定义写入文件的名字
  2. 定义写入内容格式
  3. 满足给定条件后更改写入的文件
  4. 更改写入文件时触发的 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


更多干货,欢迎关注公众号,哈尔的数据城堡

原文链接:https://yq.aliyun.com/articles/678983
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章