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

一. 概述

上一篇我們介紹了如何將數(shù)據(jù)從 mysql 拋到 kafka，這次我們就專注于利用 storm 將數(shù)據(jù)寫入到 hdfs 的過程，由于 storm 寫入 hdfs 的可定制東西有些多，我們先不從 kafka 讀取，而先自己定義一個(gè) Spout 數(shù)據(jù)充當(dāng)數(shù)據(jù)源，下章再進(jìn)行整合。這里默認(rèn)你是擁有一定的 storm 知識(shí)的基礎(chǔ)，起碼知道 Spout 和 bolt 是什么。

寫入 hdfs 可以有以下的定制策略：

  1. 自定義寫入文件的名字
  2. 定義寫入內(nèi)容格式
  3. 滿足給定條件后更改寫入的文件
  4. 更改寫入文件時(shí)觸發(fā)的 Action

本篇會(huì)先說明如何用 storm 寫入 HDFS，寫入過程一些 API 的描述，以及最后給定一個(gè)例子：

storm 每接收到 10 個(gè) Tuple 后就會(huì)改變 hdfs 寫入文件，新文件的名字就是第幾次改變。

ps：storm 版本：1.1.1 。Hadoop 版本：2.7.4 。

接下來我們首先看看 Storm 如何寫入 HDFS 。

二. Storm 寫入 HDFS

Storm 官方有提供了相應(yīng)的 API 讓我們可以使用?？梢酝ㄟ^創(chuàng)建 HdfsBolt 以及定義相應(yīng)的規(guī)則，即可寫入 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 這個(gè)插件，然后使用 maven Lifecycle 中的 package 打包。而不是用 Maven-assembly-plugin 插件進(jìn)行打包。

因?yàn)槭褂?Maven-assembly-plugin 的時(shí)候，會(huì)將所有依賴的包unpack，然后在pack，這樣就會(huì)出現(xiàn)，同樣的文件被覆蓋的情況。發(fā)布到集群上的時(shí)候就會(huì)報(bào) No FileSystem for scheme: hdfs 的錯(cuò) 。

然后是使用 HdfsBolt 寫入 Hdfs。這里來看看官方文檔中的例子吧。

// 使用 "|" 來替代 ","，來進(jìn)行字符分割
RecordFormat format = new DelimitedRecordFormat()
        .withFieldDelimiter("|");

// 每輸入 1k 后將內(nèi)容同步到 Hdfs 中
SyncPolicy syncPolicy = new CountSyncPolicy(1000);

// 當(dāng)文件大小達(dá)到 5MB ，轉(zhuǎn)換寫入文件，即寫入到一個(gè)新的文件中
FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);

//當(dāng)轉(zhuǎn)換寫入文件時(shí)，生成新文件的名字并使用
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");

到這里就結(jié)束了?？梢詫?HdfsBolt 當(dāng)作一個(gè) Storm 中特殊一些的 bolt 即可。這個(gè) bolt 的作用即使根據(jù)接收信息寫入 Hdfs。

而在新建 HdfsBolt 中，Storm 為我們提供了相當(dāng)強(qiáng)的靈活性，我們可以定義一些策略，比如當(dāng)達(dá)成某個(gè)條件的時(shí)候轉(zhuǎn)換寫入文件，新寫入文件的名字，寫入時(shí)候的分隔符等等。

如果選擇使用的話，Storm 有提供部分接口供我們使用，但如果我們覺得不夠豐富也可以自定義相應(yīng)的類。下面我們看看如何控制這些策略吧。

RecordFormat

這是一個(gè)接口，允許你自由定義接收到內(nèi)容的格式。

public interface RecordFormat extends Serializable {
    byte[] format(Tuple tuple);
}

Storm 提供了 DelimitedRecordFormat ，使用方法在上面已經(jīng)有了。這個(gè)類默認(rèn)的分割符是逗號(hào)","，而你可以通過 withFieldDelimiter 方法改變分隔符。
如果你的初始分隔符不是逗號(hào)的話，那么也可以重寫寫一個(gè)類實(shí)現(xiàn) RecordFormat 接口即可。

FileNameFormat

同樣是一個(gè)接口。

public interface FileNameFormat extends Serializable {
    void prepare(Map conf, TopologyContext topologyContext);
    String getName(long rotation, long timeStamp);
    String getPath();
}

Storm 所提供的默認(rèn)的是 org.apache.storm.hdfs.format.DefaultFileNameFormat 。默認(rèn)人使用的轉(zhuǎn)換文件名有點(diǎn)長(zhǎng)，格式是這樣的：

{prefix}{componentId}-{taskId}-{rotationNum}-{timestamp}{extension}

例如：

MyBolt-5-7-1390579837830.txt

默認(rèn)情況下，前綴是空的，擴(kuò)展標(biāo)識(shí)是".txt"。

SyncPolicy

同步策略允許你將 buffered data 緩沖到 Hdfs 文件中（從而client可以讀取數(shù)據(jù)），通過實(shí)現(xiàn)org.apache.storm.hdfs.sync.SyncPolicy 接口：

public interface SyncPolicy extends Serializable {
    boolean mark(Tuple tuple, long offset);
    void reset();
}

FileRotationPolicy

這個(gè)接口允許你控制什么情況下轉(zhuǎn)換寫入文件。

public interface FileRotationPolicy extends Serializable {
    boolean mark(Tuple tuple, long offset);
    void reset();
}

Storm 有提供三個(gè)實(shí)現(xiàn)該接口的類：

  • 最簡(jiǎn)單的就是不進(jìn)行轉(zhuǎn)換的org.apache.storm.hdfs.bolt.rotation.NoRotationPolicy ，就是什么也不干。

  • 通過文件大小觸發(fā)轉(zhuǎn)換的 org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy。

  • 通過時(shí)間條件來觸發(fā)轉(zhuǎn)換的 org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy。

如果有更加復(fù)雜的需求也可以自己定義。

RotationAction

這個(gè)主要是提供一個(gè)或多個(gè) hook ，可加可不加。主要是在觸發(fā)寫入文件轉(zhuǎn)換的時(shí)候會(huì)啟動(dòng)。

public interface RotationAction extends Serializable {
    void execute(FileSystem fileSystem, Path filePath) throws IOException;
}

三.實(shí)現(xiàn)一個(gè)例子

了解了上面的情況后，我們會(huì)實(shí)現(xiàn)一個(gè)例子，根據(jù)寫入記錄的多少來控制寫入轉(zhuǎn)換（改變寫入的文件），并且轉(zhuǎn)換后文件的名字表示當(dāng)前是第幾次轉(zhuǎn)換。

首先來看看 HdfsBolt 的內(nèi)容：

        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);

然后分別來看各個(gè)策略的類。

FileRotationPolicy

import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
import org.apache.storm.tuple.Tuple;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 計(jì)數(shù)以改變Hdfs寫入文件的位置，當(dāng)寫入10次的時(shí)候，則更改寫入文件，更改名字取決于 “TimesFileNameFormat”
 * 這個(gè)類是線程安全
 */

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;

/**
 * 決定重新寫入文件時(shí)候的名字
 * 這里會(huì)返回是第幾次轉(zhuǎn)換寫入文件，將這個(gè)第幾次做為文件名
 */
public class TimesFileNameFormat implements FileNameFormat {
    //默認(rèn)路徑
    private String path = "/storm";
    //默認(rèn)后綴
    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 ++ ;
        //返回文件名，文件名為更換寫入文件次數(shù)
        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;
/**
    當(dāng)轉(zhuǎn)換寫入文件時(shí)候調(diào)用的 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 個(gè) Tuple 后就會(huì)轉(zhuǎn)換寫入文件，新文件的名字就是第幾次轉(zhuǎn)換。

完整代碼包括一個(gè)隨機(jī)生成字符串的 Spout ，可以到我的 github 上查看。

StormHdfsDemo:https://github.com/shezhiming/StormHdfsDemo

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容