• 使用Spark清洗统计业务数据并保存到数据库中

    1、打开前面创建的项目“BigData-Etl-KongGuan”，创建一些数据库访问的工具类和辅助类。

    1）打开SpringBoot项目：BigData-Etl-KongGuan

    2）创建数据库访问的工具类和辅助类：

    com.qrsoft.etl.dao.IBaseDao数据库访问的通用类，包括创建连接、执行更新等通用操作
    com.qrsoft.etl.common.db.ConnectionPoolManager连接管理类
    com.qrsoft.etl.common.db.IConnectionPool连接池管理类接口
    com.qrsoft.etl.common.db.ConnectionPool连接池管理类接口实现类
    com.qrsoft.etl.common.db.DBbean这是外部可以配置的连接池属性 可以允许外部配置，拥有默认值
    com.qrsoft.etl.common.db.DBInitInfo初始化，模拟加载所有的配置文件
    com.qrsoft.etl.common.Constants全局常量类
    com.qrsoft.etl.util.ConfigUtil加载配置文件的工具类
    com.qrsoft.etl.util.ConfigManager配置文件管理的工具类
    com.qrsoft.etl.util.MapManager地图管理的工具类，例如：是否在矩形区域内
    • 创建com.qrsoft.etl.util.ConfigUtil类，该类是一个通用工具类，用于加载myconfig.properties配置文件，并提供了一个根据键来读取配置文件中属性值的方法readProperty(key)。
    1.  package com.qrsoft.etl.util;
    2.  
    3. import org.springframework.core.io.support.PropertiesLoaderUtils;
    4.  
    5. import java.io.IOException;
    6. import java.util.Properties;
    7.  
    8. public class ConfigUtil {
    9.     public static String readProperty(String key){
    10.         Properties properties = new Properties();
    11.         try {
    12.             properties = PropertiesLoaderUtils.loadAllProperties("myconfig.properties");
    13.         } catch (IOException e) {
    14.             e.printStackTrace();
    15.         }
    16.         return properties.get(key).toString();
    17.     }
    18. }
    • 创建com.qrsoft.etl.util.ConfigManager类，该类是一个通用工具类，用于配置文件管理，在该类的构造函数中加载config.properties配置文件，并提供了一个根据键来读取配置文件中属性值的方法getValue(key)。
    1. package com.qrsoft.etl.util;
    2.  
    3. import java.io.IOException;
    4. import java.io.InputStream;
    5. import java.util.Properties;
    6.  
    7. public class ConfigManager {
    8. 	private static ConfigManager configManager; // 声明工具类的一个私有的对象
    9. 	private static Properties properties; //声明对象
    10.  
    11. 	private ConfigManager() { //私有无参构造方法
    12. 		String configFile = "config.properties";
    13. 		properties = new Properties();
    14. 		InputStream in = ConfigManager.class.getClassLoader().getResourceAsStream(configFile);
    15. 		try {
    16. 			properties.load(in);
    17. 			in.close();
    18. 		} catch (IOException e) {
    19. 			e.printStackTrace();
    20. 		}
    21. 	}
    22.  
    23. 	public static ConfigManager getInstance() {
    24. 		if (configManager == null) {
    25. 			configManager = new ConfigManager();
    26. 		}
    27. 		return configManager;
    28. 	}
    29.  
    30. 	public String getValue(String key) {
    31. 		return properties.getProperty(key);
    32. 	}
    33. }
    • 创建com.qrsoft.etl.util.MapManager类，该类是地图管理的工具类，提供了地图操作的相关工具方法，例如：地图上某个经纬度的点是否在矩形区域内、判断是否在经纬度范围内等。
    1. package com.qrsoft.etl.util;
    2.  
    3. public class MapManager {
    4.     /**
    5.      * 是否在矩形区域内
    6.      * @Title: isInArea
    7.      * @Description: TODO()
    8.      * @param lat 测试点经度
    9.      * @param lng 测试点纬度
    10.      * @param minLat 纬度范围限制1
    11.      * @param maxLat 纬度范围限制2
    12.      * @param minLng 经度限制范围1
    13.      * @param maxLng 经度范围限制2
    14.      * @return boolean
    15.      */
    16.     public  boolean isInRectangleArea(double lat,double lng,double minLat,
    17.  double maxLat,double minLng,double maxLng){
    18.         if(this.isInRange(lat, minLat, maxLat)){//如果在纬度的范围内
    19.             if(minLng*maxLng>0){
    20.                 if(this.isInRange(lng, minLng, maxLng)){
    21.                     return true;
    22.                 }else {
    23.                     return false;
    24.                 }
    25.             }else {
    26.                 if(Math.abs(minLng)+Math.abs(maxLng)<180){
    27.                     if(this.isInRange(lng, minLng, maxLng)){
    28.                         return true;
    29.                     }else {
    30.                         return false;
    31.                     }
    32.                 }else{
    33.                     double left = Math.max(minLng, maxLng);
    34.                     double right = Math.min(minLng, maxLng);
    35.                     if(this.isInRange(lng, left, 180)||this.isInRange(lng, right,-180)){
    36.                         return true;
    37.                     }else {
    38.                         return false;
    39.                     }
    40.                 }
    41.             }
    42.         }else{
    43.             return false;
    44.         }
    45.     }
    46.     /**
    47.      * 判断是否在经纬度范围内
    48.      */
    49.     public  boolean isInRange(double point, double left,double right){
    50.         if(point>=Math.min(left, right)&&point<=Math.max(left, right)){
    51.             return true;
    52.         }else {
    53.             return false;
    54.         }
    55.     }
    56. }
    • 创建com.qrsoft.etl.common.db.DBbean类，该类中定义了外部可以配置的连接池属性 可以允许外部配置，拥有默认值。
    1. package com.qrsoft.etl.common.db;
    2.  
    3. public class DBbean {
    4.     // 连接池属性
    5.     private String driverName;
    6.     private String url;
    7.     private String userName;
    8.     private String password;
    9.     // 连接池名字
    10.     private String poolName;
    11.     private int minConnections = 10; // 空闲池，最小连接数
    12.     private int maxConnections = 300; // 空闲池，最大连接数
    13.     private int initConnections = 20;// 初始化连接数
    14.     private long connTimeOut = 1000;// 重复获得连接的频率
    15.     private int maxActiveConnections = 500;// 最大允许的连接数，和数据库对应
    16.     private long connectionTimeOut = 0;// 连接超时时间，默认20分钟
    17.     private boolean isCurrentConnection = true; // 是否获得当前连接，默认true
    18.     private boolean isCheakPool = true; // 是否定时检查连接池
    19.     private long lazyCheck = 1000 * 60 * 60;// 延迟多少时间后开始 检查
    20.     private long periodCheck = 1000 * 60 * 60;// 检查频率
    21.  
    22.     public DBbean(String driverName, String url, String userName,String password, String poolName) {
    23.         super();
    24.         this.driverName = driverName;
    25.         this.url = url;
    26.         this.userName = userName;
    27.         this.password = password;
    28.         this.poolName = poolName;
    29.     }
    30.     public DBbean() {
    31.     }
    32.     public String getDriverName() {
    33.         if (driverName == null) {
    34.             driverName = this.getDriverName() + "_" + this.getUrl();
    35.         }
    36.         return driverName;
    37.     }
    38.     public void setDriverName(String driverName) {
    39.         this.driverName = driverName;
    40.     }
    41.     public String getUrl() {
    42.         return url;
    43.     }
    44.     public void setUrl(String url) {
    45.         this.url = url;
    46.     }
    47.     public String getUserName() {
    48.         return userName;
    49.     }
    50.     public void setUserName(String userName) {
    51.         this.userName = userName;
    52.     }
    53.     public String getPassword() {
    54.         return password;
    55.     }
    56.     public void setPassword(String password) {
    57.         this.password = password;
    58.     }
    59.     public String getPoolName() {
    60.         return poolName;
    61.     }
    62.     public void setPoolName(String poolName) {
    63.         this.poolName = poolName;
    64.     }
    65.     public int getMinConnections() {
    66.         return minConnections;
    67.     }
    68.     public void setMinConnections(int minConnections) {
    69.         this.minConnections = minConnections;
    70.     }
    71.     public int getMaxConnections() {
    72.         return maxConnections;
    73.     }
    74.     public void setMaxConnections(int maxConnections) {
    75.         this.maxConnections = maxConnections;
    76.     }
    77.     public int getInitConnections() {
    78.         return initConnections;
    79.     }
    80.     public void setInitConnections(int initConnections) {
    81.         this.initConnections = initConnections;
    82.     }
    83.  
    84.     public int getMaxActiveConnections() {
    85.         return maxActiveConnections;
    86.     }
    87.     public void setMaxActiveConnections(int maxActiveConnections) {
    88.         this.maxActiveConnections = maxActiveConnections;
    89.     }
    90.     public long getConnTimeOut() {
    91.         return connTimeOut;
    92.     }
    93.     public void setConnTimeOut(long connTimeOut) {
    94.         this.connTimeOut = connTimeOut;
    95.     }
    96.     public long getConnectionTimeOut() {
    97.         return connectionTimeOut;
    98.     }
    99.     public void setConnectionTimeOut(long connectionTimeOut) {
    100.         this.connectionTimeOut = connectionTimeOut;
    101.     }
    102.     public boolean isCurrentConnection() {
    103.         return isCurrentConnection;
    104.     }
    105.     public void setCurrentConnection(boolean isCurrentConnection) {
    106.         this.isCurrentConnection = isCurrentConnection;
    107.     }
    108.     public long getLazyCheck() {
    109.         return lazyCheck;
    110.     }
    111.     public void setLazyCheck(long lazyCheck) {
    112.         this.lazyCheck = lazyCheck;
    113.     }
    114.     public long getPeriodCheck() {
    115.         return periodCheck;
    116.     }
    117.     public void setPeriodCheck(long periodCheck) {
    118.         this.periodCheck = periodCheck;
    119.     }
    120.     public boolean isCheakPool() {
    121.         return isCheakPool;
    122.     }
    123.     public void setCheakPool(boolean isCheakPool) {
    124.         this.isCheakPool = isCheakPool;
    125.     }
    126. }
    • 创建com.qrsoft.etl.common.db.DBInitInfo类，该类主要用于加载配置文件中所有的有关数据库的配置信息，初始化数据库连接对象。
    1. package com.qrsoft.etl.common.db;
    2.  
    3. import com.qrsoft.etl.util.ConfigManager;
    4.  
    5. import java.util.ArrayList;
    6. import java.util.List;
    7.  
    8. public class DBInitInfo {
    9.     // 设置注册属性
    10.     public static String DRIVER = ConfigManager.getInstance().getValue("jdbc.driver");
    11.     public static String URL = ConfigManager.getInstance().getValue("jdbc.url");
    12.     public static String USERNAME = ConfigManager.getInstance().getValue("jdbc.username");
    13.     public static String PASSWORD = ConfigManager.getInstance().getValue("jdbc.password");
    14.     public static String MinConnections = ConfigManager.getInstance().getValue("jdbc.min");
    15.     public static String MaxConnections = ConfigManager.getInstance().getValue("jdbc.max");
    16.  
    17.     public static List beans = null;
    18.     static {
    19.         beans = new ArrayList();
    20.         // 这里数据 可以从xml 等配置文件进行获取，为了测试，这里直接写死了
    21.         DBbean beanMysql = new DBbean();
    22.         beanMysql.setDriverName(DRIVER);
    23.         beanMysql.setUrl(URL);
    24.         beanMysql.setUserName(USERNAME);
    25.         beanMysql.setPassword(PASSWORD);
    26.  
    27.         beanMysql.setMinConnections(Integer.parseInt(MinConnections));
    28.         beanMysql.setMaxConnections(Integer.parseInt(MaxConnections));
    29.  
    30.         beanMysql.setPoolName("pool");
    31.         beans.add(beanMysql);
    32.     }
    33. }
    • 创建com.qrsoft.etl.common.db.IConnectionPool接口，该接口中定义了连接池操作类的接口方法，如：获得连接、回收连接、销毁清空、查看连接池活动状态、定时检查连接池等接口方法。
    1. package com.qrsoft.etl.common.db;
    2.  
    3. import java.sql.Connection;
    4. import java.sql.SQLException;
    5.  
    6. public interface IConnectionPool {
    7.     // 获得连接
    8.     public Connection getConnection();
    9.     // 获得当前连接
    10.     public Connection getCurrentConnecton();
    11.     // 回收连接
    12.     public void releaseConn(Connection conn) throws SQLException;
    13.     // 销毁清空
    14.     public void destroy();
    15.     // 连接池是活动状态
    16.     public boolean isActive();
    17.     // 定时器，检查连接池
    18.     public void cheackPool();
    19. }
    • 创建com.qrsoft.etl.common.db.ConnectionPool类，该类实现了IConnectionPool接口，并实现了接口中定义的方法。
    1. package com.qrsoft.etl.common.db;
    2.  
    3. import java.sql.Connection;
    4. import java.sql.DriverManager;
    5. import java.sql.SQLException;
    6. import java.util.List;
    7. import java.util.Timer;
    8. import java.util.TimerTask;
    9. import java.util.Vector;
    10.  
    11. public class ConnectionPool implements IConnectionPool {
    12.     // 连接池配置属性
    13.     private DBbean dbBean;
    14.     private boolean isActive = false; // 连接池活动状态
    15.     private int contActive = 0;// 记录创建的总的连接数
    16.     // 空闲连接
    17.     private List<Connection> freeConnection = new Vector<Connection>();
    18.     // 活动连接
    19.     private List<Connection> activeConnection = new Vector<Connection>();
    20.     // 将线程和连接绑定，保证事务能统一执行
    21.     private static ThreadLocal<Connection> threadLocal = new ThreadLocal<Connection>();
    22.     public ConnectionPool(DBbean dbBean) {
    23.         super();
    24.         this.dbBean = dbBean;
    25.         init();
    26.         cheackPool();
    27.     }
    28.     // 初始化
    29.     public void init() {
    30.         try {
    31.             Class.forName(dbBean.getDriverName());
    32.             for (int i = 0; i < dbBean.getInitConnections(); i++) {
    33.                 Connection conn;
    34.                 conn = newConnection();
    35.                 // 初始化最小连接数
    36.                 if (conn != null) {
    37.                     freeConnection.add(conn);
    38.                     contActive++;
    39.                 }
    40.             }
    41.             isActive = true;
    42.         } catch (ClassNotFoundException e) {
    43.             e.printStackTrace();
    44.         } catch (SQLException e) {
    45.             e.printStackTrace();
    46.         }
    47.     }
    48.     // 获得当前连接
    49.     public Connection getCurrentConnecton() {
    50.         // 默认线程里面取
    51.         Connection conn = threadLocal.get();
    52.         if (!isValid(conn)) {
    53.             conn = getConnection();
    54.         }
    55.         return conn;
    56.     }
    57.     // 获得连接
    58.     public synchronized Connection getConnection() {
    59.         Connection conn = null;
    60.         try {
    61.             // 判断是否超过最大连接数限制
    62.             if (contActive < this.dbBean.getMaxActiveConnections()) {
    63.                 if (freeConnection.size() > 0) {
    64.                     conn = freeConnection.get(0);
    65.                     if (conn != null) {
    66.                         threadLocal.set(conn);
    67.                     }
    68.                     freeConnection.remove(0);
    69.                 } else {
    70.                     conn = newConnection();
    71.                 }
    72.             } else {
    73.                 // 继续获得连接,直到从新获得连接
    74.                 wait(this.dbBean.getConnTimeOut());
    75.                 conn = getConnection();
    76.             }
    77.             if (isValid(conn)) {
    78.                 activeConnection.add(conn);
    79.                 contActive++;
    80.             }
    81.         } catch (SQLException e) {
    82.             e.printStackTrace();
    83.         } catch (ClassNotFoundException e) {
    84.             e.printStackTrace();
    85.         } catch (InterruptedException e) {
    86.             e.printStackTrace();
    87.         }
    88.         return conn;
    89.     }
    90.     // 获得新连接
    91.     private synchronized Connection newConnection() throws ClassNotFoundException, SQLException {
    92.         Connection conn = null;
    93.         if (dbBean != null) {
    94.             Class.forName(dbBean.getDriverName());
    95.             conn = DriverManager.getConnection(dbBean.getUrl(), dbBean.getUserName(), dbBean.getPassword());
    96.         }
    97.         return conn;
    98.     }
    99.     // 释放连接
    100.     public synchronized void releaseConn(Connection conn) throws SQLException {
    101.         if (isValid(conn)&& !(freeConnection.size() > dbBean.getMaxConnections())) {
    102.             freeConnection.add(conn);
    103.             activeConnection.remove(conn);
    104.             contActive--;
    105.             threadLocal.remove();
    106.             // 唤醒所有正待等待的线程，去抢连接
    107.             notifyAll();
    108.         }
    109.     }
    110.  
    111.     // 判断连接是否可用
    112.     private boolean isValid(Connection conn) {
    113.         try {
    114.             if (conn == null || conn.isClosed()) {
    115.                 return false;
    116.             }
    117.         } catch (SQLException e) {
    118.             e.printStackTrace();
    119.         }
    120.         return true;
    121.     }
    122.     // 销毁连接池
    123.     public synchronized void destroy() {
    124.         for (Connection conn : freeConnection) {
    125.             try {
    126.                 if (isValid(conn)) {
    127.                     conn.close();
    128.                 }
    129.             } catch (SQLException e) {
    130.                 e.printStackTrace();
    131.             }
    132.         }
    133.         for (Connection conn : activeConnection) {
    134.             try {
    135.                 if (isValid(conn)) {
    136.                     conn.close();
    137.                 }
    138.             } catch (SQLException e) {
    139.                 e.printStackTrace();
    140.             }
    141.         }
    142.         isActive = false;
    143.         contActive = 0;
    144.     }
    145.     // 连接池状态
    146.     public boolean isActive() {
    147.         return isActive;
    148.     }
    149.     // 定时检查连接池情况
    150.     public void cheackPool() {
    151.         if (dbBean.isCheakPool()) {
    152.             new Timer().schedule(new TimerTask() {
    153.                 @Override
    154.                 public void run() {
    155.                     // 1.对线程里面的连接状态
    156.                     // 2.连接池最小 最大连接数
    157.                     // 3.其他状态进行检查，因为这里还需要写几个线程管理的类，暂时就不添加了
    158.                     System.out.println("空线池连接数：" + freeConnection.size());
    159.                     System.out.println("活动连接数：：" + activeConnection.size());
    160.                     System.out.println("总的连接数：" + contActive);
    161.                 }
    162.             }, dbBean.getLazyCheck(), dbBean.getPeriodCheck());
    163.         }
    164.     }
    165. }
    • 创建com.qrsoft.etl.common.db.ConnectionPoolManager类，该类为数据库的连接管理类，是一个通用的固定写法，通过连接池管理数据库连接，提供了获取单例模式连接的实现，还提供了根据连接池名字获得连接的方法，以及关闭连接、清空连接池等方法。
    1. package com.qrsoft.etl.common.db;
    2.  
    3. import java.sql.Connection;
    4. import java.sql.SQLException;
    5. import java.util.Hashtable;
    6. /**
    7.  * 连接管理类
    8.  */
    9. public class ConnectionPoolManager {
    10.     // 连接池存放
    11.     public Hashtable<String, IConnectionPool> pools = new Hashtable<String, IConnectionPool>();
    12.     // 初始化
    13.     private ConnectionPoolManager() {
    14.         init();
    15.     }
    16.     // 单例实现
    17.     public static ConnectionPoolManager getInstance() {
    18.         return Singtonle.instance;
    19.     }
    20.     private static class Singtonle {
    21.         private static ConnectionPoolManager instance = new ConnectionPoolManager();
    22.     }
    23.     // 初始化所有的连接池
    24.     public void init() {
    25.         for (int i = 0; i < DBInitInfo.beans.size(); i++) {
    26.             DBbean bean = DBInitInfo.beans.get(i);
    27.             ConnectionPool pool = new ConnectionPool(bean);
    28.             if (pool != null) {
    29.                 pools.put(bean.getPoolName(), pool);
    30.                 System.out.println("Info:Init connection successed ->"
    31.                         + bean.getPoolName());
    32.             }
    33.         }
    34.     }
    35.     // 获得连接,根据连接池名字 获得连接
    36.     public Connection getConnection(String poolName) {
    37.         Connection conn = null;
    38.         if (pools.size() > 0 && pools.containsKey(poolName)) {
    39.             conn = getPool(poolName).getConnection();
    40.             try {
    41.                 conn.setAutoCommit(false);
    42.             } catch (SQLException e) {
    43.                 e.printStackTrace();
    44.             }
    45.         } else {
    46.             System.out.println(
    47.                     "Error:Can't find this connecion pool ->" + poolName);
    48.         }
    49.         return conn;
    50.     }
    51.     // 关闭，回收连接
    52.     public void close(String poolName, Connection conn) {
    53.         IConnectionPool pool = getPool(poolName);
    54.         try {
    55.             if (pool != null) {
    56.                 pool.releaseConn(conn);
    57.             }
    58.         } catch (SQLException e) {
    59.             System.out.println("连接池已经销毁");
    60.             e.printStackTrace();
    61.         }
    62.     }
    63.     // 清空连接池
    64.     public void destroy(String poolName) {
    65.         IConnectionPool pool = getPool(poolName);
    66.         if (pool != null) {
    67.             pool.destroy();
    68.         }
    69.     }
    70.     // 获得连接池
    71.     public IConnectionPool getPool(String poolName) {
    72.         IConnectionPool pool = null;
    73.         if (pools.size() > 0) {
    74.             pool = pools.get(poolName);
    75.         }
    76.         return pool;
    77.     }
    78. }
    • 创建com.qrsoft.etl.dao.IBaseDao类，该类主要用于数据库的通用操作，如创建连接、执行更新等操作。
    1. package com.qrsoft.etl.dao;
    2.  
    3. import com.qrsoft.etl.common.db.ConnectionPoolManager;
    4. import com.qrsoft.etl.common.db.IConnectionPool;
    5. import org.slf4j.Logger;
    6. import org.slf4j.LoggerFactory;
    7.  
    8. import java.sql.*;
    9. import java.util.ArrayList;
    10. import java.util.HashMap;
    11. import java.util.List;
    12. import java.util.Map;
    13.  
    14. public class IBaseDao {
    15.  
    16.     private final static Logger logger = LoggerFactory.getLogger(IBaseDao.class);
    17.     public ResultSet rs = null;
    18.     private IConnectionPool pool= ConnectionPoolManager.getInstance().getPool("pool");
    19.     private Connection conn = getConnection();
    20.     // 定义sql语句的执行对象
    21.     private PreparedStatement pstmt;
    22.     public Connection getConnection(){
    23.         Connection conn = null;
    24.         if(pool != null && pool.isActive()){
    25.             conn = pool.getConnection();
    26.         }
    27.         return conn;
    28.     }
    29.     public Connection getCurrentConnection(){
    30.         Connection conn = null;
    31.         if(pool != null && pool.isActive()){
    32.             conn = pool.getCurrentConnecton();
    33.         }
    34.         return conn;
    35.     }
    36.     /**
    37.      * 查询
    38.      */
    39.     public ResultSet execute(String sql, Object params[]) {
    40.         try {
    41.             conn.setAutoCommit(false);
    42.             pstmt = conn.prepareStatement(sql);
    43.             for (int i = 0; i < params.length; i++) {
    44.                 pstmt.setObject(i + 1, params[i]);
    45.             }
    46.             rs = pstmt.executeQuery();
    47.             pool.releaseConn(conn);
    48.         } catch (SQLException e) {
    49.             e.printStackTrace();
    50.             logger.info("查询失败！", e.getMessage());
    51.         }
    52.         return rs;
    53.     }
    54.     /**
    55.      * 更新
    56.      */
    57.     public boolean update(String sql, Object params[]) throws SQLException {
    58.         boolean flag = false;
    59.         try {
    60.             conn.setAutoCommit(false);
    61.             pstmt = conn.prepareStatement(sql);
    62.             for (int i = 0; i < params.length; i++) {
    63.                 pstmt.setObject(i + 1, params[i]);
    64.             }
    65.             int i = pstmt.executeUpdate();
    66.             if (i > 0){
    67.                 flag = true;
    68.             }else{
    69.                 flag = false;
    70.             }
    71.             pool.releaseConn(conn);
    72.             conn.commit();
    73.         } catch (SQLException e) {
    74.             conn.rollback();
    75.             e.printStackTrace();
    76.             logger.info("更新失败！", e.getMessage());
    77.         }
    78.         return flag;
    79.     }
    80.     /**
    81.      * 更新一个
    82.      */
    83.     public boolean updateOne(String sql) throws SQLException {
    84.         boolean flag = false;
    85.         try {
    86.             pstmt = conn.prepareStatement(sql);
    87.             int i = pstmt.executeUpdate();
    88.             if (i > 0){
    89.                 flag = true;
    90.             } else{
    91.                 flag = false;
    92.             }
    93.             pool.releaseConn(conn);
    94.             conn.commit();
    95.         } catch (SQLException e) {
    96.             conn.rollback();
    97.             e.printStackTrace();
    98.             logger.info("更新失败！", e.getMessage());
    99.         }
    100.         return flag;
    101.     }
    102.     /**
    103.      * 执行查询操作
    104.      */
    105.     public List<Map<String, Object>> findResult(String sql, List params) throws SQLException {
    106.         List<Map<String, Object>> list = new ArrayList>();
    107.         int index = 1;
    108.         pstmt = conn.prepareStatement(sql);
    109.         if (params != null && !params.isEmpty()) {
    110.             for (int i = 0; i < params.size(); i++) {
    111.                 pstmt.setObject(index++, params.get(i));
    112.             }
    113.         }
    114.         rs = pstmt.executeQuery();
    115.         ResultSetMetaData metaData = rs.getMetaData();
    116.         int cols_len = metaData.getColumnCount();
    117.         while (rs.next()) {
    118.             Map<String, Object> map = new HashMap<String, Object>();
    119.             for (int i = 0; i < cols_len; i++) {
    120.                 String cols_name = metaData.getColumnName(i + 1);
    121.                 Object cols_value = rs.getObject(cols_name);
    122.                 if (cols_value == null) {
    123.                     cols_value = "";
    124.                 }
    125.                 map.put(cols_name, cols_value);
    126.             }
    127.             list.add(map);
    128.         }
    129.         return list;
    130.     }
    131.  
    132.     public Boolean getBool(ResultSet airRs){
    133.         boolean bool = false;
    134.         Integer no = 0;
    135.         try {
    136.             while (airRs.next()) {
    137.                 no = airRs.getInt(1);
    138.                 if (no>0) {
    139.                     bool = true;
    140.                 }
    141.             }
    142.         } catch (SQLException e) {
    143.             bool = false;
    144.             e.printStackTrace();
    145.         }
    146.         return bool;
    147.     }
    148. }
    • 创建com.qrsoft.etl.common.Constants类，该类为全局常量类，定义了相关的常量，使用的时候直接通过Constants类调用，即方便修改，又可以避免手写时写错。
    1. package com.qrsoft.etl.common;
    2.  
    3. public class Constants {
    4. 	//间隔时间（10分钟）
    5. 	public final static int INTERVAL_TIME_10MIN = 10*60*1000;
    6. 	//间隔时间（5分钟）
    7. 	public final static int INTERVAL_TIME_5MIN = 5*60*1000;
    8. 	//间隔时间（1分钟）
    9. 	public final static int INTERVAL_TIME_1MIN = 60*1000;
    10. 	//间隔时间（30秒）
    11. 	public final static int INTERVAL_TIME_30SEC = 30*1000;
    12. 	//间隔时间（10秒）
    13. 	public final static int INTERVAL_TIME_10SEC = 10*1000;
    14. 	//间隔时间（10秒）
    15. 	public final static int INTERVAL_TIME_5SEC = 5*1000;
    16. 	//每分钟读取条数
    17. 	public final static int READ_COUNT = 10;
    18.  
    19. 	//kg_airport
    20. 	public final static String TABLE_AIRPORT = "kg_airport";
    21.  
    22. 	//kg_airlinecompany
    23. 	public final static String TABLE_AIRLINECOMPANY = "kg_airlinecompany";
    24.  
    25. 	//kg_PlanData计划数据
    26. 	public final static String TASK_PlANDATA = "task_PlanData";
    27. 	public final static String TABLE_PlANDATA = "Kg_PlanData";
    28. 	public final static String FAMILY_PlANDATA = "ReportHome";
    29. 	public final static String COLUMN_PlANDATA = "EXECUTE_DATE";
    30.  
    31. 	//kg_MultiRadarData综合航迹数据
    32. 	public final static String TASK_RADAR = "task_Radar";
    33. 	public final static String TABLE_RADAR = "Kg_MultiRadarData";
    34. 	public final static String FAMILY_RADAR = "RadarHome";
    35. 	public final static String COLUMN_RADAR = "SEND_RADAR_TIME";
    36.  
    37. 	//kg_AFTN报文数据
    38. 	public final static String TASK_AFTN = "task_Aftn";
    39. 	public final static String TABLE_AFTN = "Kg_AFTN";
    40. 	public final static String FAMILY_AFTN = "AFTNHome";
    41. 	public final static String COLUMN_AFTN = "EXECUTE_DATE";
    42.  
    43. 	//Kg_ATCDutyInfo管制值班人员数据
    44. 	public final static String TASK_ATCDUTY = "task_ATCDuty";
    45. 	public final static String TABLE_ATCDUTY = "Kg_ATCDutyInfo";
    46. 	public final static String FAMILY_ATCDUTY = "ATCDutyHome";
    47. 	public final static String COLUMN_ATCDUTY = "SEND_TIME";
    48.  
    49. 	//Kg_WarnFlightHistory航班指令告警数据
    50. 	public final static String TASK_WARNFLIGHT = "task_WarnFlight";
    51. 	public final static String TABLE_WARNFLIGHT = "Kg_WarnFlightHistory";
    52. 	public final static String FAMILY_WARNFLIGHT = "WarnFlightHome";
    53. 	public final static String COLUMN_WARNFLIGHT = "GJ_DATE";
    54.  
    55. 	//Kg_WarnSimilarHistory相似航班号告警数据
    56. 	public final static String TASK_WARNSIMILAR = "task_WarnSimilar";
    57. 	public final static String TABLE_WARNSIMILAR = "Kg_WarnSimilarHistory";
    58. 	public final static String FAMILY_WARNSIMILAR = "WarnSimilarHome";
    59. 	public final static String COLUMN_WARNSIMILAR = "GJ_DATE";
    60.  
    61.     //Kg_ATC扇区信息
    62.     public final static String TASK_ATC = "task_ATC";
    63.     public final static String TABLE_ATC = "Kg_ATC";
    64.     public final static String FAMILY_ATC = "ATCHome";
    65.     public final static String COLUMN_ATC = "EXECUTE_DATE";
    66.  
    67. 	//Kg_CallSaturation 扇区通话饱和度信息
    68. 	public final static String TASK_CALLSATURATION = "task_CallSaturation";
    69. 	public final static String TABLE_CALLSATURATION = "Kg_CallSaturation";
    70. 	public final static String FAMILY_CALLSATURATION = "SaturationHome";
    71. 	public final static String COLUMN_CALLSATURATION = "SEND_TIME";
    72. }

    2、创建com/qrsoft/etl/spark/SparkStreamingApplication.java类，在该类中配置Kafka和Spark的执行参数，然后使用Spark进行业务数据处理，在代码中会根据不同的Topic进入不同的分支，进行数据的处理。

    • 创建SparkStreamingApplication.java类，设置Zookeeper的brokers和zkServers，设置需要监听的Topic List
    1. package com.qrsoft.etl.spark;
    2.  
    3. import com.qrsoft.etl.common.Constants;
    4. import com.qrsoft.etl.util.ConfigUtil;
    5. import net.sf.json.JSONObject;
    6. import org.apache.kafka.clients.consumer.ConsumerConfig;
    7. import org.apache.kafka.clients.consumer.ConsumerRecord;
    8. import org.apache.kafka.common.serialization.StringDeserializer;
    9. import org.apache.spark.SparkConf;
    10. import org.apache.spark.api.java.JavaSparkContext;
    11. import org.apache.spark.streaming.Durations;
    12. import org.apache.spark.streaming.api.java.JavaInputDStream;
    13. import org.apache.spark.streaming.api.java.JavaStreamingContext;
    14. import org.apache.spark.streaming.kafka010.ConsumerStrategies;
    15. import org.apache.spark.streaming.kafka010.KafkaUtils;
    16. import org.apache.spark.streaming.kafka010.LocationStrategies;
    17. import org.springframework.stereotype.Component;
    18.  
    19. import java.io.Serializable;
    20. import java.util.Arrays;
    21. import java.util.Collection;
    22. import java.util.HashMap;
    23. import java.util.Map;
    24. import java.util.regex.Pattern;
    25.  
    26. @Component
    27. public class SparkStreamingApplication implements Serializable {
    28. 	static final Pattern SPACE = Pattern.compile(" ");
    29.     // 多个以逗号隔开
    30.     static String brokers = ConfigUtil.readProperty("brokers");
    31.     static String zkserver = ConfigUtil.readProperty("zkserver");
    32.  
    33.     // 消费者组名称
    34.     static String groupId = "spark_etl";
    35.     // topic列表
    36.     static String topicsStr = Constants.TASK_RADAR;
    37.     static String[] topicsStrs = {
    38.             Constants.TASK_PlANDATA,
    39.             Constants.TASK_WARNFLIGHT,
    40.             Constants.TASK_ATCDUTY,
    41.             Constants.TASK_WARNSIMILAR,
    42.             Constants.TASK_AFTN,
    43.             Constants.TASK_ATC,
    44.             Constants.TASK_CALLSATURATION,
    45.             Constants.TASK_RADAR
    46.     };
    47.     /**
    48.      * 启动Spark读取、清洗数据
    49.      */
    50.     public void SparkEtlStart() {
    51. 		// ... 在此处添加代码 ...
    52. 	}
    53. }
    • 在SparkEtlStart()方法内添加代码，配置Spark和Kafka参数，创建StreamingContext

    创建StreamingContext，设置拉取流的时间，准备读取Kafka数据。本地开发时Spark配置使用local[*]方式，设置成本地运行模式，放到集群中运行时需要修改为Yarn模式。

    1. SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("qst-etl");
    2. //反压机制
    3. conf.set("spark.streaming.backpressure.enabled", "true");
    4. conf.set("allowMultipleContexts", "true");
    5. JavaSparkContext sc = new JavaSparkContext(conf);
    6. // 获取jssc 以及设置获取流的时间
    7. JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(10));
    8. jssc.sparkContext().setLogLevel("WARN");
    9. // Kafka 参数配置
    10. Map<String, Object> kafkaParams = new HashMap<>();
    11. kafkaParams.put("zookeeper.connect", zkserver);
    12. kafkaParams.put("bootstrap.servers", brokers);
    13. kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);     //指定了KafkaConsumershuyu 哪一个消费者群组
    14. kafkaParams.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    15. kafkaParams.put("key.deserializer", StringDeserializer.class);
    16. kafkaParams.put("value.deserializer", StringDeserializer.class);
    17.  
    18.  kafkaParams.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
    19.  kafkaParams.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");   //读取Kafka最新的一条
    20.  kafkaParams.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
    21.  
    22.  Collection<String> topics = Arrays.asList(topicsStrs);
    • 拉取Kafka数据流
    1. // 获取流
    2. JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(jssc,
    3.                 LocationStrategies.PreferConsistent(),
    4.                 ConsumerStrategies.Subscribe(topics, kafkaParams));
    • 解析Kafka数据流：解析流，对流进行循环处理，首先把流区分Topic，解析流中的value，其次根据不同Topic进入不同的分支，进行处理
    1.  stream.foreachRDD(rdd -> {
    2.             rdd.foreach(t -> {
    3.                 String topName = t.topic();
    4.                 JSONObject jsonObject = new JSONObject();
    5.                 String taskRadar = "";
    6.                 if(topName.equals(Constants.TASK_RADAR)){
    7.                     taskRadar = t.value();
    8.                 }else{
    9.                     jsonObject = JSONObject.parseObject(t.value());
    10.                 }
    11.                 SparkUtil sparkUtil = new SparkUtil();
    12.                 try {
    13.                     switch (topName) {
    14.                         case Constants.TASK_RADAR:
    15. //                            sparkUtil.TaskRadar(taskRadar);
    16.                             sparkUtil.TaskRadarStr(taskRadar);
    17.                             break;
    18.                         case Constants.TASK_PlANDATA:
    19.                             sparkUtil.TaskPlanData(jsonObject);
    20.                             break;
    21.                         case Constants.TASK_WARNFLIGHT:
    22.                             sparkUtil.TaskWarnfLight(jsonObject);
    23.                             break;
    24.                         case Constants.TASK_ATCDUTY:
    25.                             sparkUtil.TaskAtcduty(jsonObject);
    26.                             break;
    27.                         case Constants.TASK_WARNSIMILAR:
    28.                             sparkUtil.TaskWarnsimilar(jsonObject);
    29.                             break;
    30.                         case Constants.TASK_AFTN:
    31.                             sparkUtil.TaskAftn(jsonObject);
    32.                             break;
    33.                         case Constants.TASK_ATC:
    34.                             sparkUtil.TaskAtc(jsonObject);
    35.                             break;
    36.                         case Constants.TASK_CALLSATURATION:
    37.                             sparkUtil.TaskCallsaturation(jsonObject);
    38.                             break;
    39.                     }
    40.                 } catch (Exception e) {
    41.                     System.out.println(e);
    42.                 }
    43.                 //return Arrays.asList(SPACE.split(t.value())).iterator();
    44.             });
    45.         });
    • 启动Spark，进行业务处理
    1. // 打印结果
    2. //warns.print();
    3. try {
    4.      // 启动
    5.      jssc.start();
    6.      jssc.awaitTermination();
    7. } catch (InterruptedException e) {
    8.      e.printStackTrace();
    9. }

    3、将清洗后的数据保存到数据库中

    • 在上一步的代码中，解析Kafka数据流时，首先把流区分Topic，解析流中的value，其次根据不同Topic进入不同的分支，进行处理。例如：
    1.  case Constants.TASK_PlANDATA:
    2.       sparkUtil.TaskPlanData(jsonObject);
    3.       break;

    该分支是处理机场起降数据，这里会用到一个类SparkUtil.java，该类中定义了处理不同Topic数据的方法，其中sparkUtil.TaskPlanData(jsonObject)就是处理机场起降数据对应的方法。

    主要任务是：对起降信息进行统计和更新、对航线信息进行统计和更新：

    1）首先判断是否有该机场航班起降的统计信息，如果数据库中没有该机场数据，则在数据库中插入；如果有则根据条件进行更新数据；

    2）对于航线信息也是如此，如果数据库中没有相应的航线数据，则在数据库中插入；否则根据条件进行更新。

    • 编写com.qrsoft.etl.spark.SparkUtil类，代码位置src/main/java/com/qrsoft/etl/spark/SparkUtil.java，在该类中添加一个方法TaskPlanData(jsonObject)，用于处理“机场起降数据”对应的Topic中的数据。
    1. package com.qrsoft.etl.spark;
    2.  
    3. import com.alibaba.fastjson.JSONObject;
    4. import com.qrsoft.etl.dao.PlanDataDao;
    5. import org.slf4j.Logger;
    6. import org.slf4j.LoggerFactory;
    7. import org.springframework.stereotype.Component;
    8.  
    9. import java.text.SimpleDateFormat;
    10. import java.util.Date;
    11.  
    12. @Component
    13. public class SparkUtil {
    14.     private final static Logger logger = LoggerFactory.getLogger(SparkUtil.class);
    15.     // 初始化扇区
    16.     private static Double[] sectionG = {38.716066,42.297914,114.648477,128.759203};
    17.     private static Double[] sectionK = {35.519796,38.716066,114.648477,128.759203};
    18.     private static Double[] sectionE = {32.519796,35.519796,114.648477,128.759203};
    19.  
    20.     /**
    21.      * 业务处理
    22.      * @param jsonObject  机场起降数据
    23.      * @throws Exception
    24.      */
    25.     public void TaskPlanData(JSONObject jsonObject) throws Exception {
    26.         //起飞机场
    27.         String adep = jsonObject.getString("ADEP");
    28.         //降落机场
    29.         String ades = jsonObject.getString("ADES");
    30.         //操作数据库，统计和更新机场航班数
    31.         operationDB(adep);
    32.         operationDB(ades);
    33.         //航班号
    34.         String acid = jsonObject.getString("ACID");
    35.         //操作数据库，统计和更新航线信息
    36.         operationDBBOLT(adep, ades, acid);
    37.     }
    38.     /**
    39.      * 操作数据库（对航班起降数进行统计或更新）
    40.      * @param code “起飞机场”或“降落机场”
    41.      */
    42.     public void operationDB(String code) {
    43.         //根据机场代码获取目前数据库中已存在的航班数
    44.         PlanDataDao pDao = new PlanDataDao();
    45.         boolean bool;
    46.         try {
    47.             bool = pDao.isExistThisAir(code);
    48.             if (bool) {
    49.                 //存在，在原来基础上+1,修改数据库中该机场的航班数
    50.                 pDao.updateAnAirMsg(code);
    51.             } else {
    52.                 //不存在，在统计表中创建该机场的航班数（默认为1）
    53.                 pDao.createAnAirMsg(code);
    54.             }
    55.         } catch (Exception e) {
    56.             e.printStackTrace();
    57.         }
    58.     }
    59.  
    60.     /**
    61.      * 操作数据库（对航线进行统计或更新）
    62.      * @param adep 起飞机场
    63.      * @param ades 降落机场
    64.      * @param acid 航班号
    65.      */
    66.     public void operationDBBOLT(String adep, String ades, String acid) {
    67.         boolean bool;
    68.         PlanDataDao pDao = new PlanDataDao();
    69.         if (pDao.isDomesticThisLine(adep) && pDao.isDomesticThisLine(ades)) {
    70.             bool = pDao.isExistThisLine(acid);
    71.             if (bool) {
    72.                 pDao.updateAnLineMsg(acid);
    73.             } else {
    74.                 pDao.createAnLineMsg(acid, adep, ades);
    75.             }
    76.         }
    77.     }
    78.     // ... ...
    79.     // ... 其他方法。因为当前要实现的是“机场起降数据”，所以其他可以只有方法体，没有方法实现及返回值。 ...
    80.     // ... ...
    81.     public void TaskRadarStr(String taskRadar) { }
    82.     public void TaskWarnfLight(JSONObject jsonObject) { }
    83.     public void TaskAtcduty(JSONObject jsonObject) { }
    84.     public void TaskWarnsimilar(JSONObject jsonObject) { }
    85.     public void TaskAftn(JSONObject jsonObject) { }
    86.     public void TaskAtc(JSONObject jsonObject) { }
    87.     public void TaskCallsaturation(JSONObject jsonObject) { }
    88.     // ... ... 
    89. }
    • 在上面的代码中会用到一个com.qrsoft.etl.dao.PlanDataDao类，代码位置src/main/java/com/qrsoft/etl/dao/PlanDataDao.java。该类是一个数据库操作类，会根据业务逻辑进行实际的数据库的操作，如增、删、改、查等。
    1. package com.qrsoft.etl.dao;
    2.  
    3. import org.slf4j.Logger;
    4. import org.slf4j.LoggerFactory;
    5.  
    6. import java.sql.ResultSet;
    7. import java.sql.SQLException;
    8.  
    9.  
    10. public class PlanDataDao extends IBaseDao {
    11.  
    12.     private final static Logger logger = LoggerFactory.getLogger(PlanDataDao.class);
    13.     
    14.     // ... ...
    15.     // ... 添加方法实现 ... 
    16.     // ... ...
    17. }
    • 对于“处理机场起降数据”会涉及到以下的方法：

    查询该机场是否在国内：

    1. public boolean isDomesticThisLine(String code4){
    2.         String sql = " SELECT COUNT(*) from airport_longlat where code4 ='"+code4+"';";
    3.         Object[] params = {};
    4.         ResultSet comRs = this.execute(sql, params);
    5.         return getBool(comRs);
    6. }

    根据机场代码查询是否有该机场的统计信息：

    1. public boolean isExistThisAir(String code) {
    2. 	String sql = " SELECT COUNT(*) from airport_number where flightcode='"+code+"';";
    3. 	Object[] params = {};
    4. 	ResultSet airRs = this.execute(sql, params);
    5. 	return getBool(airRs);
    6. }

    如果根据机场代码查询，有该机场的统计信息，则在数据库中更新机场的起降航班数：

    1. public void updateAnAirMsg(String code) {
    2. 	String sql = "update airport_number set count=count+'1' where flightcode='"+code+"'; ";
    3. 	Object[] params = {};
    4. 	try {
    5. 		this.update(sql, params);
    6. 	} catch (SQLException e) {
    7. 		logger.info("修改指定机场的统计信息（统计数在原来基础上+1）失败！  " + code);
    8. 		e.printStackTrace();
    9. 	}
    10. }

    如果根据机场代码查询，有该机场的统计信息，则在统计表中创建该机场的航班数（默认为1）：

    1. public void createAnAirMsg(String code) {
    2. 	String sql = "insert into airport_number (flightcode,cname,count) values ('"+code+"',(select airport_cname from kg_airport where AIRPORT_CODE4 = '"+code+"'),'1');";
    3. 	Object[] params = {};
    4. 	try {
    5. 		this.update(sql, params);
    6. 	} catch (SQLException e) {
    7. 		logger.info("创建新机场的统计信息失败！  " + code);
    8. 		e.printStackTrace();
    9. 	}
    10. }

    根据航班号查询是否有该航线存在：

    1. public boolean isExistThisLine(String acid){
    2.     String sql = " SELECT COUNT(*) from airline_number where acid ='"+acid+"';";
    3.     Object[] params = {};
    4.     ResultSet comRs = this.execute(sql, params);
    5.     return getBool(comRs);
    6. }

    根据航班号查询，有该航线统计信息，则在统计表中修改指定航线的统计信息（统计数在原来的基础上+1）：

    1. public void updateAnLineMsg(String acid) {
    2.     String sql = "update airline_number set count=count+1 where acid='"+acid+"';";
    3.     Object[] params = {};
    4.     try {
    5.         this.update(sql, params);
    6.     } catch (SQLException e) {
    7.         logger.info("修改指定航线统计信息（统计数在原来基础上+1）失败! 航班号：" + acid);
    8.         e.printStackTrace();
    9.     }
    10. }

    根据航班号查询，没该航线统计信息，则创建新航线的统计信息：

    1. public void createAnLineMsg(String acid,String aDEP,String aDES) {
    2.     String sql = "insert into airline_number (acid,adepcode,adescode,adepname,adesname,adeplong,adeplat,adeslong,adeslat,count) values ('"+acid+"','"+aDEP+"','"+aDES+"',(select airport_cname from kg_airport where airport_code4 = '"+aDEP+"'),(select airport_cname from kg_airport where airport_code4 = '"+aDES+"'),(select longitude from airport_longlat where code4 = '"+aDEP+"'),(select latitude from airport_longlat where code4 = '"+aDEP+"'),(select longitude from airport_longlat where code4 = '"+aDES+"'),(select latitude from airport_longlat where code4 = '"+aDES+"'),'1') ;";
    3.     Object[] params = {};
    4.     try {
    5.         this.update(sql, params);
    6.     } catch (SQLException e) {
    7.         logger.info("创建新航线的统计信息失败！ 航班号：" + acid);
    8.         e.printStackTrace();
    9.     }
    10. }

    该PlanDataDao.java类是一个普通的数据库操作类，非常简单，主要是根据我们的业务需求对数据进行相应的操作。这里会涉及到前面我们创建的一些数据库访问的工具类和辅助类：

    com.qrsoft.etl.dao.IBaseDao数据库访问的通用类，包括创建连接、执行更新等通用操作
    com.qrsoft.etl.common.db.ConnectionPoolManager连接管理类
    com.qrsoft.etl.common.db.IConnectionPool连接池管理类接口
    com.qrsoft.etl.common.db.ConnectionPool连接池管理类接口实现类
    com.qrsoft.etl.common.db.DBbean这是外部可以配置的连接池属性 可以允许外部配置，拥有默认值
    com.qrsoft.etl.common.db.DBInitInfo初始化，模拟加载所有的配置文件
    com.qrsoft.etl.common.Constants全局常量类
    com.qrsoft.etl.util.ConfigUtil加载配置文件的工具类
    com.qrsoft.etl.util.ConfigManager配置文件管理的工具类
    com.qrsoft.etl.util.MapManager地图管理的工具类，例如：是否在矩形区域内
    • 添加或修改如下配置文件：

    1）application.yml，服务器相关配置

    1. server:
    2.   port: 8849
    3.  
    4. spring:
    5.   datasource:
    6.     driver-class-name: com.mysql.jdbc.Driver
    7.     username: root
    8.     password: 123456
    9.     url: jdbc:mysql://node3:3306/kongguan?serverTimezone=UTC
    10.   redis:
    11.     host: node3
    12.     port: 6379
    13.     database: 15
    14.  
    15. mybatis-plus:
    16.   configuration:
    17.     log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
    18.   mapper-locations: classpath:/mapper/*.xml

    2）config.properties，MySQL数据库相关配置

    1. jdbc.driver=com.mysql.jdbc.Driver
    2. jdbc.url=jdbc:mysql://node3:3306/kongguan?useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai
    3. jdbc.username=root
    4. jdbc.password=123456
    5. jdbc.min=20
    6. jdbc.max=500

    3）log4j.properties，日志文件相关配置

    1. #定义LOG输出级别
    2. #log4j.rootLogger=INFO,Console,File,stdout
    3. log4j.rootLogger=Console
    4. log4j.rootCategory = ERROR,Console
    5. #定义日志输出目的地为控制台
    6. log4j.appender.Console=org.apache.log4j.ConsoleAppender
    7. log4j.appender.Console.Target=System.out
    8. #可以灵活地指定日志输出格式，下面一行是指定具体的格式
    9. log4j.appender.Console.layout = org.apache.log4j.PatternLayout
    10. log4j.appender.Console.layout.ConversionPattern=[%c] - %m%n
    11.  
    12. #文件大小到达指定尺寸的时候产生一个新的文件
    13. # log4j.appender.File = org.apache.log4j.RollingFileAppender
    14. log4j.appender.File.Append=true
    15. log4j.appender.File = org.apache.log4j.DailyRollingFileAppender
    16. #指定输出目录
    17. log4j.appender.File.File = /home/tmp/hbase
    18. log4j.appender.File.DatePattern = '_'yyyy-MM-dd'.log'
    19. #定义文件最大大小
    20. log4j.appender.File.MaxFileSize = 10MB
    21. # 输出所以日志，如果换成DEBUG表示输出DEBUG以上级别日志
    22. log4j.appender.File.Threshold = ERROR
    23. log4j.appender.File.layout = org.apache.log4j.PatternLayout
    24. log4j.appender.File.layout.ConversionPattern =[%p] [%d{yyyy-MM-dd HH\:mm\:ss}][%c]%m%n
    25.  
    26. # mybatis日志输出
    27. log4j.logger.com.sarnath.ind.dao.IRoleDao.addPermission=TRACE
    28. log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    29. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    30. log4j.appender.stdout.layout.ConversionPattern=%5p [%t] - %m%n

    4）myconfig.properties，Zookeeper和Kafka相关配置

    1. brokers: node1:9092,node2:9092,node3:9092
    2. zkserver: node1:2181,node2:2181,node3:2181

    注意，在上面的步骤中只处理了一个Topic分支的数据，其余Topic分支，参考TASK_PlANDATA方式处理，自行实现其他XXX统计（请参考源代码）。

    4、修改项目启动文件BigDataEtlKongGuanApplication.java，内容如下：

    1. package com.qrsoft;
    2.  
    3. import com.qrsoft.etl.spark.SparkStreamingApplication;
    4. import org.springframework.boot.SpringApplication;
    5. import org.springframework.boot.autoconfigure.SpringBootApplication;
    6. import org.springframework.context.ConfigurableApplicationContext;
    7. import org.springframework.scheduling.annotation.EnableScheduling;
    8.  
    9. @SpringBootApplication
    10. @EnableScheduling
    11. public class BigDataEtlKongGuanApplication {
    12.     public static void main(String[] args) {
    13.         ConfigurableApplicationContext run = SpringApplication.run(BigDataEtlKongGuanApplication.class, args);
    14.         SparkStreamingApplication bean = run.getBean(SparkStreamingApplication.class);
    15.         bean.SparkEtlStart();
    16.     }
    17. }

    5、测试

    • 确保Hadoop、Spark、Kafka、MySQL等环境均已经启动，如果没有启动，可参考前面的安装部署任务，自行启动。
    • 启动BigData-KongGuan项目（如果没有启动）
    • 启动BigData-Etl-KongGuan项目

    • 在node3节点，进入MySQL数据库，查询 airline*相关的表是否有数据
    [root@node3 ~]# mysql -uroot -p123456
    mysql> show databases;
    mysql> use kongguan;
    mysql> select * from airline_number limit 20;

    例如：统计airline_number表，记录数是递增的。

    mysql> select count(*) from airline_number;

  • 相关阅读:
    React 18 迁移状态逻辑至 Reducer 中
    UI设计师岗位的基本职责八篇
    排队（单调队列+二分）
    学历不好，还有希望进大厂吗？
    引导滤波融合matlab
    【WPF应用30】WPF中的ListBox控件详解
    Android基础-进程间通信
    HCIA-MSTP替代技术之链路捆绑（LACP模式）
    【unity实战】Unity实现2D人物双击疾跑
    java 比较运算符
  • 原文地址：https://blog.csdn.net/2201_75642955/article/details/136752301