2. 程式人生 > >spark+phoenix 通過jdbc讀取表中的資料

spark+phoenix 通過jdbc讀取表中的資料

阿新 • • 發佈：2019-01-29

廢話不說，直接程式碼，解決燃煤之急

新增maven配置
<dependency>
   <groupId>org.apache.phoenix</groupId>
   <artifactId>phoenix-core</artifactId>
   <version>${phoenix.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.phoenix</groupId>
  <artifactId>
 
phoenix-spark</artifactId>
  <version>${phoenix.version}</version>
</dependency>
spark讀取phoenix中的資料
package com.cctsoft.spark.offline;

import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * Created with IntelliJ IDEA.
 * User: Kevin Liu
 * CreateDate: 2018/6/8 10:32
 * Description: 讀取phoenix表中資料
 */
public class FaceCrashImsiJob {
    public static void main(String[] args)  throws AnalysisException {
        // $example on:init_session$
        SparkSession spark = SparkSession
                .builder()
                .appName("Java Spark SQL basic example")
                .config("spark.some.config.option", "some-value")
                .master("local")
                .getOrCreate();
        // $example off:init_session$

        runBasicDataFrameExample(spark);


        spark.stop();
    }


    private static void runBasicDataFrameExample(SparkSession spark) throws AnalysisException {

       

    /**
     * 獲取4G資料
     * String tableName= "(select * from LTE_DATA where to_char(cap_time) >= '"+imsiDataStartTime+"' and to_char(cap_time) < '"+imsiDataEndTime+"') as LTE_DATA_FILTER";
     */
    String imsiTableName = "(select * from LTE_DATA where to_char(cap_time) >= '"+imsiDataStartTime+"' and to_char(cap_time) < '"+imsiDataEndTime+"') as LTE_DATA_FILTER";
    logger.info("imsiTableName:"+imsiTableName);
    Dataset<Row> df = spark.read().format("jdbc")
            .option("driver","org.apache.phoenix.jdbc.PhoenixDriver")
            .option("url","jdbc:phoenix:"+zookeeper+":2181")
            .option("dbtable",imsiTableName)
            .load();
    df.registerTempTable("lte_data_tmp");
    Dataset<Row> lteDataAll = spark.sql("select lte_dev_code,cap_time,imsi from lte_data_tmp order by cap_time desc");
    lteDataAll.show();



    }




}
通過spark寫入資料到phoenix
package com.cctsoft.spark.offline
import org.apache.spark.api.java.JavaRDD
import org.apache.spark.sql.Row

/**
  * Created with IntelliJ IDEA.
  * User: Kevin Liu
  * CreateDate: 2018/6/15 12:32
  * Description: 寫入phoenix表資料
  */
object TestMain {
  def main(args: Array[String]): Unit 
 
= {
    batchSaveFaceImsi(null)
  }


  def batchSaveFaceImsi(imsiRdd: JavaRDD[Row]): Unit ={
    import org.apache.phoenix.spark._

    val rdd = imsiRdd.rdd.map(x=>{
      (x.get(0).toString+":"+x.get(1).toString,x.get(0).toString,x.get(1).toString)
    }).filter(f=>f._3.toString!=null && f._3.toString!="")

    rdd.foreach(println)

    rdd.saveToPhoenix(
        "RESIDENT_TMP",
Seq("ID","DEVICE_MAC","IMSI"),
zkUrl = Some("jdbc:phoenix:abigdataclient1:2181")
    )



  }

}

相關推薦

spark+phoenix 通過jdbc讀取表中的資料

廢話不說，直接程式碼，解決燃煤之急新增maven配置<dependency> <groupId>org.apache.phoenix</groupId>

access vba 用recordset讀取表中資料的簡單方法

'strQuery是表名，查詢名等 Public Function Getrs(Byval strQuery as string) as ADODB.Recordset Dim objRs As New ADODB.Recordset on Error GoTo Er

vB程式設計VB原始碼 VB讀取EXCEL工作薄某個表中資料 ADODB.Recordset

 Private Sub Form_Load() Dim RS As ADODB.Recordset Set RS = GetExcelRs(App.Path & "\book1.xls") MsgBox RS.RecordCount MsgBox RS(0) RS

利用poi將excel表中資料讀取存入mysql資料庫（資料量比較大）

最近被老大安排了一個任務，利用程式將excle表中的資料讀取到，做處理，然後存進資料庫。接到任務的時候人是懵逼的。但是安排的任務也得硬著頭皮完成。現將做的東西記錄如下，方便以後查詢。 這個小demo的原型是在網上找的，demo連結如下 http://www.cnblogs.

map端join的實現 ，用來解決小表中資料的讀取

通過閱讀父類Mapper的原始碼，發現 setup方法是在maptask處理資料之前呼叫一次 可以用來做一些初始化工作 1、需求： 訂單資料表t_order： id date pid amount 1001 20150710 P0001 2

spark 讀取elasticsearch中資料不完整問題

使用spark讀取elasticsearch中的資料，使用es提供的api來進行， sc.esRDD("logstash").values 官方網站也是這種方式讀取資料的，但是我測試的時候有時候會出現讀取資料不完整的情況，比如本來讀取的資料是這樣的 Map(msg ->

C#.Net 4 通過對映讀取DbDataReader中的資料

前段時間比較鬱悶，想做個通用的函式，從DbDataReader中讀取所有資料。 上傳一小段程式碼，也算是拋磚引玉了。 /// <summary> /// 通過對映取DbDataReader中的資料，只取最後一條資料 /// <

spark流式讀取hdfs中資料

名詞解釋： spark streaming： 定義：一個對實時資料進行高容通量、容錯處理的流式處理系統，可以對多種資料來源進行Map、reduce和join等複雜操作，並將結果儲存到外部檔案系統、

spark streaming 通過zookeeper讀取kafka上的資料

maven 依賴如下 <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-c

Oracle刪除表中資料的時候應該注意的問題 Oracle與MySql他們一些表之間的區別

1. 刪除表的注意事項 在刪除一個表中的全部資料時，須使用TRUNCATE TABLE 表名;因為用DROP TABLE，DELETE * FROM 表名時，TABLESPACE表空間該表的佔用空間並未釋放，反覆幾次DROP，DELETE操作後，該TABLESPACE上百兆的空間就被耗光了。

Hive Shell 命令之二（表中資料的操作，出自Hive程式設計指南）

一、 互動模式： show tables; #檢視所有表名 show tables  'ad*'  #檢視以'ad'開頭的表名 set 命令 #設定變數與檢視變數； set -v #檢視所有的變數 set hive.stats.atomic #檢視hive.sta

JDBC插入表中一個數據返回其自增主鍵

Connection connection = JDBCUtils.getConnection(); //獲取一個連結 String sql ="insert into book values (null ,?)"; //資料庫book表中只有一個id自增跟書名欄位 PreparedSta

Java將資料庫表中資料匯出至Excel表格

        在我們學習和使用Java的過程中，會有匯出資料庫中的資料（或其他資料）到表格中的需求。比如你建了一個部落格網站，也寫了很多篇部落格，那麼就想把寫的所有部落格匯出到xls表格中。因為資料在伺服器可能會面臨資料丟失的風險（比如伺服器重灌系統，伺服器

SQL SERVER匯出表中資料的sql指令碼形式

CREATE proc spGenInsertSQL (@tablename varchar(256)) as begin declare @sql varchar(8000) declare @sqlValues varchar(8000) set @sql =' (' set @sqlVal

如何實現一個表字段資料變化，另一張表中資料隨著變化

資料表1（db_integrityevent）： 資料表2：（db_corporatecreditscore）： navicat觸發器寫法： 可以實現表中相關資料變化，表中其他跟隨這變化。 mysql命令列寫法： delimiter $$ for each row B

倒置線性表中資料的順序（c++）

倒置線性表中資料的順序（c++） 給LList類實現新增一個成員函式，倒置線性表中資料的順序，且演算法的執行時間為O（n） template<typename E> void LList<E>::reverse(){ Link<E> * p,*

如何對比mysql資料表的資料與excel表中資料是否重複？

一、利用excel比對 1、將mysql資料表匯出為excel 2、將表格資料與excel資料進行對比 方法一：函式COUNTIF 　　　　IF(COUNTIF(資料庫資料所在的區域,目標值所在區域)=0,”不重複”,”“) 　　　　 方法二：函式Ma

Oracle 表，表中資料不小心刪除找回的方法

一：drop 掉的表可以通過閃回來恢復 比如：   flashback table t1 to timestamp to_timestamp('2018-10-15 13:04:13','yyyy-mm-dd hh24:mi:ss'); 二：delete 的表中資料可以通

python讀取表中的json再重新寫入會出現中文變成unicode碼

有一個表中的欄位 是gbk格式，比如該欄位是comment：這是人和公園的西門 然後 將該欄位 要插入另一個表(utf8)中的某個欄位：該欄位是個json， 往該json中添 加一個新的欄位commen

mysql在把子查詢結果作為刪除表中資料的條件，mysql不允許在子查詢的同時刪除原表資料

** 目標表和原表一致 MYSQL不支援如此操作 資料庫不能邊查詢邊刪除 ** delete from push_msg_overview where id in ( select id from push_msg_overview where pus