大数据培训之代码实现

1 导入Pom依赖

<dependencies>

    <dependency>

        <groupId>org.apache.flume</groupId>

        <artifactId>flume-ng-core</artifactId>

        <version>1.7.0</version>

    </dependency>

    <dependency>

        <groupId>mysql</groupId>

        <artifactId>mysql-connector-java</artifactId>

        <version>5.1.27</version>

    </dependency>

</dependencies>

2 添加配置信息

在ClassPath下添加jdbc.properties和log4j. properties

jdbc.properties:

dbDriver=com.mysql.jdbc.Driver

dbUrl=jdbc:mysql://hadoop102:3306/mysqlsource?useUnicode=true&characterEncoding=utf-8
dbUser=root
dbPassword=000000

log4j. properties:

#——–console———–

log4j.rootLogger=info,myconsole,myfile
log4j.appender.myconsole=org.apache.log4j.ConsoleAppender
log4j.appender.myconsole.layout=org.apache.log4j.SimpleLayout
#log4j.appender.myconsole.layout.ConversionPattern =%d [%t] %-5p [%c] – %m%n

#log4j.rootLogger=error,myfile
log4j.appender.myfile=org.apache.log4j.DailyRollingFileAppender
log4j.appender.myfile.File=/tmp/flume.log
log4j.appender.myfile.layout=org.apache.log4j.PatternLayout
log4j.appender.myfile.layout.ConversionPattern =%d [%t] %-5p [%c] – %m%n

3 SQLSourceHelper

1)属性说明:

属性

说明(括号中为默认值)

runQueryDelay    

查询时间间隔(10000)

batchSize

缓存大小(100)

startFrom

查询语句开始id(0)

currentIndex

查询语句当前id,每次查询之前需要查元数据表

recordSixe

查询返回条数

table

监控的表名

columnsToSelect

查询字段(*)

customQuery

用户传入的查询语句

query

查询语句

defaultCharsetResultSet

编码格式(UTF-8)

2)方法说明:

方法

说明

SQLSourceHelper(Context context)

构造方法,初始化属性及获取JDBC连接

InitConnection(String url, String user, String pw)

获取JDBC连接

checkMandatoryProperties()

校验相关属性是否设置(实际开发中可增加内容)

buildQuery()

根据实际情况构建sql语句,返回值String

executeQuery()

执行sql语句的查询操作,返回值List<List<Object>>

getAllRows(List<List<Object>> queryResult)

将查询结果转换为String,方便后续操作

updateOffset2DB(int size)

根据每次查询结果将offset写入元数据表

execSql(String sql)

具体执行sql语句方法

getStatusDBIndex(int startFrom)

获取元数据表中的offset

queryOne(String sql)

获取元数据表中的offset实际sql语句执行方法

close()

关闭资源

3)代码分析

大数据培训课程

4)代码实现:

package com.atguigu.source;

 

import org.apache.flume.Context;

import org.apache.flume.conf.ConfigurationException;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

 

import java.io.IOException;

import java.sql.*;

import java.text.ParseException;

import java.util.ArrayList;

import java.util.List;

import java.util.Properties;

 

public class SQLSourceHelper {

 

    private static final Logger LOG = LoggerFactory.getLogger(SQLSourceHelper.class);

 

    private int runQueryDelay, //两次查询的时间间隔

            startFrom,            //开始id

            currentIndex,         //当前id

            recordSixe = 0,      //每次查询返回结果的条数

            maxRow;                //每次查询的最大条数

 

    private String table,       //要操作的表

            columnsToSelect,     //用户传入的查询的列

            customQuery,          //用户传入的查询语句

            query,                 //构建的查询语句

            defaultCharsetResultSet;//编码集

 

    //上下文,用来获取配置文件

    private Context context;

 

    //为定义的变量赋值(默认值),可在flume任务的配置文件中修改

    private static final int DEFAULT_QUERY_DELAY = 10000;

    private static final int DEFAULT_START_VALUE = 0;

    private static final int DEFAULT_MAX_ROWS = 2000;

    private static final String DEFAULT_COLUMNS_SELECT = “*”;

    private static final String DEFAULT_CHARSET_RESULTSET = “UTF-8”;

 

    private static Connection conn = null;

    private static PreparedStatement ps = null;

    private static String connectionURL, connectionUserName, connectionPassword;

 

    //加载静态资源

static {

 

        Properties p = new Properties();

 

        try {

            p.load(SQLSourceHelper.class.getClassLoader().getResourceAsStream(“jdbc.properties”));

            connectionURL = p.getProperty(“dbUrl”);

            connectionUserName = p.getProperty(“dbUser”);

            connectionPassword = p.getProperty(“dbPassword”);

            Class.forName(p.getProperty(“dbDriver”));

 

        } catch (IOException | ClassNotFoundException e) {

            LOG.error(e.toString());

        }

    }

 

    //获取JDBC连接

    private static Connection InitConnection(String url, String user, String pw) {

        try {

 

            Connection conn = DriverManager.getConnection(url, user, pw);

 

            if (conn == null)

                throw new SQLException();

 

            return conn;

 

        } catch (SQLException e) {

            e.printStackTrace();

        }

 

        return null;

    }

 

    //构造方法

SQLSourceHelper(Context context) throws ParseException {

 

        //初始化上下文

        this.context = context;

 

        //有默认值参数:获取flume任务配置文件中的参数,读不到的采用默认值

        this.columnsToSelect = context.getString(“columns.to.select”, DEFAULT_COLUMNS_SELECT);

 

        this.runQueryDelay = context.getInteger(“run.query.delay”, DEFAULT_QUERY_DELAY);

 

        this.startFrom = context.getInteger(“start.from”, DEFAULT_START_VALUE);

 

        this.defaultCharsetResultSet = context.getString(“default.charset.resultset”, DEFAULT_CHARSET_RESULTSET);

 

        //无默认值参数:获取flume任务配置文件中的参数

        this.table = context.getString(“table”);

        this.customQuery = context.getString(“custom.query”);

 

        connectionURL = context.getString(“connection.url”);

 

        connectionUserName = context.getString(“connection.user”);

 

        connectionPassword = context.getString(“connection.password”);

 

        conn = InitConnection(connectionURL, connectionUserName, connectionPassword);

 

        //校验相应的配置信息,如果没有默认值的参数也没赋值,抛出异常

        checkMandatoryProperties();

 

        //获取当前的id

        currentIndex = getStatusDBIndex(startFrom);

 

        //构建查询语句

        query = buildQuery();

    }

 

    //校验相应的配置信息(表,查询语句以及数据库连接的参数)

private void checkMandatoryProperties() {

 

        if (table == null) {

            throw new ConfigurationException(“property table not set”);

        }

 

        if (connectionURL == null) {

            throw new ConfigurationException(“connection.url property not set”);

        }

 

        if (connectionUserName == null) {

            throw new ConfigurationException(“connection.user property not set”);

        }

 

        if (connectionPassword == null) {

            throw new ConfigurationException(“connection.password property not set”);

        }

    }

 

    //构建sql语句

private String buildQuery() {

 

        String sql = “”;

 

        //获取当前id

        currentIndex = getStatusDBIndex(startFrom);

        LOG.info(currentIndex + “”);

 

        if (customQuery == null) {

            sql = “SELECT ” + columnsToSelect + ” FROM ” + table;

        } else {

            sql = customQuery;

        }

 

        StringBuilder execSql = new StringBuilder(sql);

 

        //以id作为offset

        if (!sql.contains(“where”)) {

            execSql.append(” where “);

            execSql.append(“id”).append(“>”).append(currentIndex);

 

            return execSql.toString();

        } else {

            int length = execSql.toString().length();

 

            return execSql.toString().substring(0, length – String.valueOf(currentIndex).length()) + currentIndex;

        }

    }

 

    //执行查询

List<List<Object>> executeQuery() {

 

        try {

            //每次执行查询时都要重新生成sql,因为id不同

            customQuery = buildQuery();

 

            //存放结果的集合

            List<List<Object>> results = new ArrayList<>();

 

            if (ps == null) {

                //

                ps = conn.prepareStatement(customQuery);

            }

 

            ResultSet result = ps.executeQuery(customQuery);

 

            while (result.next()) {

 

                //存放一条数据的集合(多个列)

                List<Object> row = new ArrayList<>();

 

                //将返回结果放入集合

                for (int i = 1; i <= result.getMetaData().getColumnCount(); i++) {

                    row.add(result.getObject(i));

                }

 

                results.add(row);

            }

 

            LOG.info(“execSql:” + customQuery + “\nresultSize:” + results.size());

 

            return results;

        } catch (SQLException e) {

            LOG.error(e.toString());

 

            // 重新连接

            conn = InitConnection(connectionURL, connectionUserName, connectionPassword);

 

        }

 

        return null;

    }

 

    //将结果集转化为字符串,每一条数据是一个list集合,将每一个小的list集合转化为字符串

List<String> getAllRows(List<List<Object>> queryResult) {

 

        List<String> allRows = new ArrayList<>();

 

        if (queryResult == null || queryResult.isEmpty())

            return allRows;

 

        StringBuilder row = new StringBuilder();

 

        for (List<Object> rawRow : queryResult) {

 

            Object value = null;

 

            for (Object aRawRow : rawRow) {

 

                value = aRawRow;

 

                if (value == null) {

                    row.append(“,”);

                } else {

                    row.append(aRawRow.toString()).append(“,”);

                }

            }

 

            allRows.add(row.toString());

            row = new StringBuilder();

        }

 

        return allRows;

    }

 

    //更新offset元数据状态,每次返回结果集后调用。必须记录每次查询的offset值,为程序中断续跑数据时使用,以id为offset

    void updateOffset2DB(int size) {

        //以source_tab做为KEY,如果不存在则插入,存在则更新(每个源表对应一条记录)

        String sql = “insert into flume_meta(source_tab,currentIndex) VALUES(‘”

                + this.table

                + “‘,'” + (recordSixe += size)

                + “‘) on DUPLICATE key update source_tab=values(source_tab),currentIndex=values(currentIndex)”;

 

        LOG.info(“updateStatus Sql:” + sql);

 

        execSql(sql);

    }

 

    //执行sql语句

private void execSql(String sql) {

 

        try {

            ps = conn.prepareStatement(sql);

 

            LOG.info(“exec::” + sql);

 

            ps.execute();

        } catch (SQLException e) {

            e.printStackTrace();

        }

    }

 

    //获取当前id的offset

private Integer getStatusDBIndex(int startFrom) {

 

        //从flume_meta表中查询出当前的id是多少

        String dbIndex = queryOne(“select currentIndex from flume_meta where source_tab='” + table + “‘”);

 

        if (dbIndex != null) {

            return Integer.parseInt(dbIndex);

        }

 

        //如果没有数据,则说明是第一次查询或者数据表中还没有存入数据,返回最初传入的值

        return startFrom;

    }

 

    //查询一条数据的执行语句(当前id)

private String queryOne(String sql) {

 

        ResultSet result = null;

 

        try {

            ps = conn.prepareStatement(sql);

            result = ps.executeQuery();

 

            while (result.next()) {

                return result.getString(1);

            }

        } catch (SQLException e) {

            e.printStackTrace();

        }

 

        return null;

    }

 

    //关闭相关资源

void close() {

 

        try {

            ps.close();

            conn.close();

        } catch (SQLException e) {

            e.printStackTrace();

        }

    }

 

    int getCurrentIndex() {

        return currentIndex;

    }

 

    void setCurrentIndex(int newValue) {

        currentIndex = newValue;

    }

 

    int getRunQueryDelay() {

        return runQueryDelay;

    }

 

    String getQuery() {

        return query;

    }

 

    String getConnectionURL() {

        return connectionURL;

    }

 

    private boolean isCustomQuerySet() {

        return (customQuery != null);

    }

 

    Context getContext() {

        return context;

    }

 

    public String getConnectionUserName() {

        return connectionUserName;

    }

 

    public String getConnectionPassword() {

        return connectionPassword;

    }

 

    String getDefaultCharsetResultSet() {

        return defaultCharsetResultSet;

    }

}

4 MySQLSource

代码实现:

package com.atguigu.source;

 

import org.apache.flume.Context;

import org.apache.flume.Event;

import org.apache.flume.EventDeliveryException;

import org.apache.flume.PollableSource;

import org.apache.flume.conf.Configurable;

import org.apache.flume.event.SimpleEvent;

import org.apache.flume.source.AbstractSource;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

 

import java.text.ParseException;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

 

public class SQLSource extends AbstractSource implements Configurable, PollableSource {

 

    //打印日志

private static final Logger LOG = LoggerFactory.getLogger(SQLSource.class);

 

    //定义sqlHelper

    private SQLSourceHelper sqlSourceHelper;

 

 

    @Override

    public long getBackOffSleepIncrement() {

        return 0;

    }

 

    @Override

    public long getMaxBackOffSleepInterval() {

        return 0;

    }

 

    @Override

public void configure(Context context) {

 

        try {

            //初始化

            sqlSourceHelper = new SQLSourceHelper(context);

        } catch (ParseException e) {

            e.printStackTrace();

        }

    }

 

    @Override

public Status process() throws EventDeliveryException {

 

        try {

            //查询数据表

            List<List<Object>> result = sqlSourceHelper.executeQuery();

 

            //存放event的集合

            List<Event> events = new ArrayList<>();

 

            //存放event头集合

            HashMap<String, String> header = new HashMap<>();

 

            //如果有返回数据,则将数据封装为event

            if (!result.isEmpty()) {

 

                List<String> allRows = sqlSourceHelper.getAllRows(result);

 

                Event event = null;

 

                for (String row : allRows) {

                    event = new SimpleEvent();

                    event.setBody(row.getBytes());

                    event.setHeaders(header);

                    events.add(event);

                }

 

                //将event写入channel

                this.getChannelProcessor().processEventBatch(events);

 

                //更新数据表中的offset信息

                sqlSourceHelper.updateOffset2DB(result.size());

            }

 

            //等待时长

            Thread.sleep(sqlSourceHelper.getRunQueryDelay());

 

            return Status.READY;

        } catch (InterruptedException e) {

            LOG.error(“Error procesing row”, e);

 

            return Status.BACKOFF;

        }

    }

 

    @Override

public synchronized void stop() {

 

        LOG.info(“Stopping sql source {} …”, getName());

 

        try {

            //关闭资源

            sqlSourceHelper.close();

        } finally {

            super.stop();

        }

    }

}

想要了解跟多关于大数据培训机构课程内容欢迎关注尚硅谷大数据培训机构,尚硅谷除了这些技术文章外还有免费的高质量大数据培训机构课程视频供广大学员下载学习


上一篇:
下一篇:
关于尚硅谷
教育理念
名师团队
学员心声
资源下载
视频下载
资料下载
工具下载
加入我们
招聘岗位
岗位介绍
招贤纳师
联系我们
电话:010-56253825
邮箱:info@atguigu.com
地址:北京市昌平区宏福科技园综合楼6层(北京校区)

 深圳市宝安区西部硅谷大厦B座C区一层(深圳校区)

上海市松江区谷阳北路166号大江商厦6层(上海校区)