乡下人产国偷v产偷v自拍,国产午夜片在线观看,婷婷成人亚洲综合国产麻豆,久久综合给合久久狠狠狠9

  • <output id="e9wm2"></output>
    <s id="e9wm2"><nobr id="e9wm2"><ins id="e9wm2"></ins></nobr></s>

    • 分享

      通過SparkStreaming的foreachRDD把處理后的數(shù)據(jù)寫入外部存儲系統(tǒng)中

       看風(fēng)景D人 2019-02-24

      1、技術(shù)實現(xiàn)解析
      2、實現(xiàn)實戰(zhàn)

      關(guān)鍵部分代碼:

              resultRowRDD.foreachPartition( partitionOfRecords => {
                // ConnectionPool is a static, lazily initialized pool of connections
      
                  val connection = ConnectionPool.getConnection()
                  partitionOfRecords.foreach(record => {
                    val sql = "insert into categorytop3(category, item, click_count) values ('" + record.getAs("category") + "','" +
                      record.getAs("item") + "'," + record.getAs("click_count") + ")"
                    val stmt = connection.createStatement()
                    stmt.executeUpdate(sql)
                  })
                  ConnectionPool.returnConnection(connection)
      
              })

      ConnectionPool.java

      package com.tom.spark.sparkstreaming;
      
      import java.sql.Connection;
      import java.sql.DriverManager;
      import java.util.LinkedList;
      
      public class ConnectionPool {
          private static LinkedList<Connection> connectionQueue;
      
          static {
              try {
                  Class.forName("com.mysql.jdbc.Driver");
              } catch (ClassNotFoundException e) {
                  e.printStackTrace();
              }
          }
          public synchronized static Connection getConnection() {
              try {
                  if(connectionQueue == null) {
                      connectionQueue = new LinkedList<Connection>();
                      for(int i = 0; i < 5; i++) {
                          Connection conn = DriverManager.getConnection(
                                  "jdbc:mysql://Master:3306/sparkstreaming",
                                  "root",
                                  "778899");
                          connectionQueue.push(conn);
                      }
                  }
              } catch(Exception e) {
                  e.printStackTrace();
              }
              return connectionQueue.poll();
          }
          public static void returnConnection(Connection conn) {connectionQueue.push(conn);}
      }

        本站是提供個人知識管理的網(wǎng)絡(luò)存儲空間,所有內(nèi)容均由用戶發(fā)布,不代表本站觀點。請注意甄別內(nèi)容中的聯(lián)系方式、誘導(dǎo)購買等信息,謹(jǐn)防詐騙。如發(fā)現(xiàn)有害或侵權(quán)內(nèi)容,請點擊一鍵舉報。
        轉(zhuǎn)藏 分享 獻花(0

        0條評論

        發(fā)表

        請遵守用戶 評論公約

        類似文章 更多