Mysql 流增量写入 Hdfs（二） --Storm + hdfs 的流式处理-低调大师

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

2018-12-11 480

一. 概述

上一篇我们介绍了如何将数据从 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

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

微信关注我们

原文链接：https://yq.aliyun.com/articles/678983

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

MaxCompute表设计最佳实践

MaxCompute表设计最佳实践产生大量小文件的操作 MaxCompute表的小文件会影响存储和计算性能，因此我们先介绍下什么样的操作会产生大量小文件，从而在做表设计的时候考虑避开此类操作。使用MaxCompute Tunnel SDK上传数据，上传过程中，每commit一次就会产生一个文件。这时每个文件过小(比如几K)，并且频繁上传(比如5秒上传)一次，则一小时就会产生720个小文件，一天就会产生17280个小文件。使用MaxCompute Tunnel SDK上传数据，create了session但是没有upload数据直接做了 commit，产生大量空目录(服务侧等同于小文件)。使用MaxCompute Console命令行工具Tunnel命令上传时，将本地大文件切分过小，导致上传后产生文件数过多，文件过小。通过Da

2018-12-12

613

列族管理在java中HColumnDescriptor代表列族,但是已经过时了,新代替的是ColumnFamilyDescriptorBuilder来创建列族描述符 ... ColumnFamilyDescriptor newc = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("newc")).build(); Admin admin = connection.getAdmin(); TableName tableName = TableName.valueOf("test"); admin.addColumnFamily(tableName,newc); ... Admin 是管理功能最重要的部分,HBaseAdmin是Admin接口的实现类,并实现了自动关闭资源的接口下面是一些比较重要的属性设置数据生存时间即数据插入后存在表中的时间,到时间时候该数据将被清理掉,是以秒来计时的 ColumnFamilyDescriptor newc = ColumnFamilyDescriptorBuilder.newBuil...

2018-12-12

582

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。