flink 自定义 mysql source

mysql source

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object FlinkDemo06_CustomSource_mysql {
    def main(args: Array[String]): Unit = {
        //1 环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //2 流对象
        import org.apache.flink.api.scala._
        val dStream: DataStream[Flight] = env.addSource(new MySqlSource)
        //3 计算 统计次数
        dStream.map(k=>(1,1)).keyBy(0).sum(1).print()
        //4 执行
        env.execute("custom mysql source")
    }
    
    class MySqlSource extends RichSourceFunction[Flight] {
        private var connection: Connection = null
        private var ps: PreparedStatement = null
        
        override def open(parameters: Configuration): Unit = {
            val driver = "com.mysql.jdbc.Driver"
            val url = "jdbc:mysql://localhost:3306/test_db"
            val username = "root"
            Class.forName(driver)
            connection = DriverManager.getConnection(url, username, "123456")
            val sql = "select id,avgTicketPrice,cancelled,carrier,dest,destAirportID,origin,originAirportID from flight"
            ps = connection.prepareStatement(sql)
        }
        
        override def close(): Unit = {
            if (connection != null) {
                connection.close()
            }
            if (ps != null) {
                ps.close()
            }
        }
        
        override def run(ctx: SourceFunction.SourceContext[Flight]): Unit = {
            //获取结果集，遍历并输出，关闭结果集
            val rs = ps.executeQuery()
            while(rs.next()) {
                val flight = Flight(rs.getFloat("avgTicketPrice"),rs.getString("cancelled"),rs.getString("carrier"),rs.getString("dest"),rs.getString("destAirportID"),rs.getString("origin"),rs.getString("originAirportID"))
                ctx.collect(flight)
            }
            rs.close()
        }
        
        override def cancel(): Unit = {}
    }
    
    case class Flight(avgTicketPrice: Float, cancelled: String, carrier: String, dest: String, destAirportID: String, origin: String, originAirportID: String)    
}

数据准备

import java.io.{BufferedReader, FileReader}
import java.sql.{Connection, DriverManager, PreparedStatement}

import org.slf4j.LoggerFactory

object FlinkDemo06_DataHelp {
    val logger = LoggerFactory.getLogger(FlinkDemo06_DataHelp.getClass)
    
    val url = "jdbc:mysql://localhost:3306/test_db"
    val driver = "com.mysql.jdbc.Driver"
    val username = "root"
    
    case class Flight(avgTicketPrice: String, cancelled: String, carrier: String, dest: String, destAirportID: String, origin: String, originAirportID: String)
    
    def main(args: Array[String]): Unit = {
        Class.forName(driver)
        var conn: Connection = null
        var stat: PreparedStatement = null
        val br = new BufferedReader(new FileReader("I:\projectImplement\dataWareHouse\test-es\data\630data.csv"))
        var line: String = null
        try {
            conn = DriverManager.getConnection(url, username, "123456")
            val sql =
                s"""
                   |insert into flight
                   | (id,avgTicketPrice,cancelled,carrier,dest,destAirportID,origin,originAirportID)
                   |values
                   | (null,?,?,?,?,?,?,?)
                   |""".stripMargin
            stat = conn.prepareStatement(sql)
            
            while ((line = br.readLine()) != null) {
                val vals = line.split(",")
                stat.setFloat(1, vals(0).toFloat)
                stat.setString(2, vals(1))
                stat.setString(3, vals(2))
                stat.setString(4, vals(3))
                stat.setString(5, vals(4))
                stat.setString(6, vals(5))
                stat.setString(7, vals(6))
                val count = stat.executeUpdate()
                if (count != 1) logger.error("插入失败")
                stat.clearParameters()
                
                Thread.sleep(10000L)
            }
        } catch {
            case e: Exception => {
                logger.error(e.toString)
            }
        } finally {
            try {
                if (br != null) {
                    br.close()
                }
                if (stat != null) {
                    stat.close()
                }
                if (conn != null) {
                    conn.close()
                }
            } catch {
                case e: Exception => logger.info(e.toString)
            }
        }
    }
}

/*
create table flight (
id int auto_increment,
avgTicketPrice float,
cancelled varchar(5),
carrier varchar(100),
dest varchar(100),
destAirportID varchar(100),
origin varchar(100),
originAirportID varchar(100),
primary key(id)
);
 */

642.5951538085938 ,false,Logstash Airways,OR Tambo International Airport,JNB,Al Maktoum International Airport ,DWC
328.17108154296875,false,ES-Air,Licenciado Benito Juarez International Airport ,AICM ,Copenhagen Kastrup Airport,CPH
774.8056030273438 ,false,Kibana Airlines ,Stockholm-Arlanda Airport,ARN,San Diego International Airport,SAN
650.2644653320312 ,false,JetBeats,Ukrainka Air Base ,XHBU ,Adolfo Suarez Madrid— Barajas Airport ,MAD
227.354248046875,false,Kibana Airlines ,Winnipeg / James Armstrong Richardson International Airport,YWG,Winnipeg / James Armstrong Richardson International Airport ,YWG
575.6364135742188 ,false,Logstash Airways,Gimpo International Airport ,GMP,Genoa Cristoforo Colombo Airport ,GE01 
619.7527465820312 ,true ,ES-Air,Tokyo Haneda International Airport ,HND,Shanghai Hongqiao International Airport ,SHA
612.101318359375,false,Logstash Airways,Reno Tahoe International Airport,RNO,Dubai International Airport,DXB
998.6478881835938 ,false,Kibana Airlines ,Xi'an Xianyang International Airport ,XIY,Kempegowda International Airport ,BLR
573.07666015625,false,Kibana Airlines ,Kansai International Airport,KIX,Beijing Capital International Airport ,PEK