Flink流计算编程--Flink sink to Oracle

前端之家收集整理的这篇文章主要介绍了Flink流计算编程--Flink sink to Oracle前端之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。

关于Flink connectors,Flink 1.1提供了许多内置的第三方连接器,这些connectors包括

Apache Kafka (sink/source)
Elasticsearch (sink)
Elasticsearch 2x (sink)
Hadoop FileSystem (sink)
RabbitMQ (sink/source)
Amazon Kinesis Streams (sink/source)
Twitter Streaming API (source)
Apache NiFi (sink/source)
Apache Cassandra (sink)
Redis (sink)

可以看到,第三方软件中,可以作为source的软件有:
Apache Kafka、RabbitMQ、Twitter Streaming API和Apache NiFi。
可以作为sink的软件包括Apache Kafka、Apache Cassandra、Redis等。

除了Flink内置支持的这些第三方软件之外,Flink也提供了自定义的source以及自定义的Sink。

2、关于Sink to JDBC

Flink的DataStream在计算完成后,就要将结果输出,目前除了上述提到的Kafka、Redis等之外,Flink也提供了其他几种方式:

writeAsText() / TextOutputFormat: 将元素按照行输出,每行当做一个字符串

writeAsCsv(...) / CsvOutputFormat: 将每行的元组按照特定的格式划分,然后输出到csv

print() / printToErr() :标准输出错误输出。也是把每行按照字符串方式输出到taskmanager的out文件

writeUsingOutputFormat() / FileOutputFormat:自定义文件输出

writeToSocket:根据序列化的sckame将元素写入socket

addSink:通过invoke方法自定义sink

其中,addSink就是我们这里要说的“自定义Flink sink”。既然是自定义,我们就可以将DataStream输出到JDBC,例如MysqL、Oracle。这在很多时候都很用。

一般情况下,我们通常会将DataStream sink到类似于Redis这种内存数据库,同时也会将结果入库,作为以后分析使用,例如sink到MysqL或oracle等JDBC。

Flink内置并没有支持JDBC,除了要覆写addSink方法外,我们还需要导入JDBC相应的依赖。
在大多数Maven管理的项目中,我们通常需要手动导入JDBC的依赖包。

3、Maven导入Oracle的依赖包

(1)获取oracle的lib包
这里要根据目标数据库的版本下载,也可以去数据库服务器中的“{ORACLE_HOME}\jdbc\lib\ojdbc.jar”,例如oracle12.1.0.2.0,此时对应的jar包是ojdbc6.jar。
(2)手动安装ojdbc6.jar到本地maven的repository

mvn install:install-file -DgroupId=com.oracle -DartifactId=ojdbc6 -Dversion=12.1.0.2.0 -Dpackaging=jar -Dfile=ojdbc6.jar

通常,cmd后先cd到刚才我们获得的ojdbc6.jar所在的路径,然后执行上边的命令即可。
(3)在pom.xml中添加引用

<!-- 添加oracle jdbc driver -->
        <dependency>
            <groupId>com.oracle</groupId>
            <artifactId>ojdbc6</artifactId>
            <version>12.1.0.2.0</version>
        </dependency>

4、Flink自定义Sink to Oracle

这里Flink对DataStrem提供了一个addSink方法,我们自定义一个类实现SinkFunction或者RichSinkFunction即可,例如:

txSumNew.addSink(new TX1MinuteSinkToOracle)

这个新的对象TX1MinuteSinkToOracle要继承RichSinkFunction,因为对数据库的insert操作,我们只需要一次性建立1个session即可,不需要每次insert都建立一个连接,因此在RichSinkFunction的open方法中建立连接。这样,就可以open一次,insert多次了。

这里采用java.sql类库建立连接,但是在连接之前,我们要告诉Flink我们准备使用的JdbcDriver是oracle Driver,如下:

Class.forName ("oracle.jdbc.OracleDriver")

之后,就可以建立连接(这里的IP要自己连接自己的服务器,s1是服务名,用户名和密码自己设定):

conn = DriverManager.getConnection("jdbc:oracle:thin:@<IP>:1521:s1","s1","s1")
      val sql = "insert into flink_tx(tx_date,minute,window_start_time,window_end_time,code,minute_volume,minute_turnover,minute_size,minute_avg_volume,sum_size,minute_volume_super,minute_volume_big,minute_volume_middle,minute_volume_small," +
        "sum_volume_super,sum_volume_big,sum_volume_middle,sum_volume_small,minute_vwap,sum_vwap,vwap_sd) values (?,?,?)"
      ps = conn.prepareStatement(sql)

完整的代码如下:

object TX1MinuteSinkToOracle {

  // *****************************************************************************
  // open()中只执行一次,开始时执行;invoke根据input进行sql执行;close()最后时关闭
  // *****************************************************************************
  class TX1MinuteSinkToOracle extends RichSinkFunction[TX1MinSliding]{

    var conn : Connection = null
    var ps: PreparedStatement = null

    val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")


    override def open(parameters: Configuration): Unit = {


      Class.forName ("oracle.jdbc.OracleDriver")

      conn = DriverManager.getConnection("jdbc:oracle:thin:@<IP>:1521:s1",?)"
      ps = conn.prepareStatement(sql)
    }


    override def invoke(in: TX1MinSliding): Unit = {

      try {

        val avg_volume_1Min = BigDecimal.valueOf(in.volume_1Min)./(BigDecimal.valueOf(in.size_1Min.toDouble)).setScale(4,BigDecimal.RoundingMode.HALF_UP)

        ps.setString(1,in.date)//时间
        ps.setString(2,in.minute)//分钟
        ps.setString(3,in.window_start_time)//窗口开始范围
        ps.setString(4,in.window_end_time)//窗口结束范围
        ps.setString(5,in.code)//股票代码
        ps.setLong(6,in.volume_1Min)//分钟成交量
        ps.setDouble(7,BigDecimal.valueOf(in.turnover_1Min).setScale(1,BigDecimal.RoundingMode.HALF_UP).toDouble)//分钟成交额
        ps.setInt(8,in.size_1Min)//分钟交易笔数
        ps.setDouble(9,avg_volume_1Min.toDouble)//分钟平均交易量
        ps.setInt(10,in.sum_size)//累计交易笔数
        ps.setDouble(11,in.volume_super_1Min.toDouble)//分钟-成交量(特大户)
        ps.setDouble(12,in.volume_big_1Min)//分钟-成交量(大户)
        ps.setDouble(13,in.volume_middle_1Min.toDouble)//分钟-成交量(中户)
        ps.setDouble(14,in.volume_small_1Min.toDouble)//分钟-成交量(散户)
        ps.setDouble(15,in.volume_super_sum.toDouble)//累计成交量(特大户)
        ps.setDouble(16,in.volume_big_sum.toDouble)//累计成交量(大户)
        ps.setDouble(17,in.volume_middle_sum.toDouble)//累计成交量(中户)
        ps.setDouble(18,in.volume_small_sum.toDouble)//累计成交量(散户)
        ps.setDouble(19,in.vwap_1Min.setScale(4,BigDecimal.RoundingMode.HALF_UP).toDouble)//分钟VWAP
        ps.setDouble(20,in.vwap_sum.setScale(4,BigDecimal.RoundingMode.HALF_UP).toDouble)//累计VWAP
        ps.setDouble(21,in.vwap_sd.setScale(4,BigDecimal.RoundingMode.HALF_UP).toDouble)//VWAP标准差

        ps.executeUpdate()

      }catch{
        case e : Exception => println(e.getMessage)
      }
    }

    override def close(): Unit = {
      if (ps != null) {
        ps.close()
      }
      if(conn != null){
        conn.close()
      }
    }

  }
}

关于Flink的addSink,我们可以看到源码如下:
\flink-streaming-java\src\main\java\org\apache\flink\streaming\api\functions\sink\下,有RichSinkFunction.java与SinkFunction.java。

@Public
public abstract class RichSinkFunction<IN> extends AbstractRichFunction implements SinkFunction<IN> {

    private static final long serialVersionUID = 1L;

    public abstract void invoke(IN value) throws Exception;

}

5、引用

DataStream–Data Sinks
Streaming Connectors
RichSinkFunction.java

猜你在找的Oracle相关文章