Apache SparkでJavaのApache Phoenixを読み書きする方法

Question

phoenix（完全なテーブルとクエリを使用）からDataFrameとDataset（Spark 2.0）を読み込み、DataFrameとDataset （Spark 2.0で）phoenixへ、Apache Sparkでjavaで。 javaにはこれに関する文書化された例はありません。

また可能な場合はphoenixConfigurationUtilを使用して入力クラスと入力クエリを設定し、sparkContextからnewAPIHadoopRDDを読み込み、別の方法でsqlContext.read().foramt("jdbc").options(pass a map with configuration keys like driver,url,dbtable).load()を使用してもう1つの方法sqlContext.read().format("org.apache.phoenix.spark").option(pass a map with configuration keys like url,table).load()を使用して読み取ります。

検索中に、Spark 1.6のdataFramesに関するその他の質問でこれらの方法を見つけましたが、例は完全ではありませんでしたが、これらのメソッドはビット単位でのみ存在していました。 Spark 2.0のサンプルを見つけることができなかった

Answer 1

このページhttps://github.com/apache/phoenix/tree/master/phoenix-sparkには、PhoenixテーブルをRDDやDataFrameなどのサンプルとして読み込む方法が含まれています。データフレームとしてテーブルをロードする。例えば

：

import org.apache.spark.SparkContext 
import org.apache.spark.sql.SQLContext 
import org.apache.phoenix.spark._ 

val sc = new SparkContext("local", "phoenix-test") 
val sqlContext = new SQLContext(sc) 

val df = sqlContext.load(
    "org.apache.phoenix.spark", 
    Map("table" -> "TABLE1", "zkUrl" -> "phoenix-server:2181") 
) 

df 
    .filter(df("COL1") === "test_row_1" && df("ID") === 1L) 
    .select(df("ID")) 
    .show 

以下要旨URLは、Javaを使用して完全な例です。

https://gist.github.com/mravi/444afe7f49821819c987

Answer 2

Scalaで鳳凰

import org.apache.spark.SparkConf; 
import org.apache.spark.SparkContext; 
import org.apache.spark.api.java.JavaSparkContext; 
import org.apache.spark.sql.DataFrame; 
import org.apache.spark.sql.SQLContext; 

import com.google.common.collect.ImmutableMap; 

import java.io.Serializable; 

public class SparkConnection implements Serializable { 

    public static void main(String args[]) { 
     SparkConf sparkConf = new SparkConf(); 
     sparkConf.setAppName("spark-phoenix-df"); 
     sparkConf.setMaster("local[*]"); 
     JavaSparkContext sc = new JavaSparkContext(sparkConf); 
     SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc); 
     DataFrame fromPhx = sqlContext.read().format("jdbc") 
       .options(ImmutableMap.of("driver", "org.apache.phoenix.jdbc.PhoenixDriver", "url", 
         "jdbc:phoenix:ZK_QUORUM:2181:/hbase-secure", "dbtable", "TABLE1")) 
       .load(); 
     fromPhx.write().format("org.apache.phoenix.spark").mode(SaveMode.Overwrite) 
     .options(ImmutableMap.of("driver", "org.apache.phoenix.jdbc.PhoenixDriver","zkUrl", 
       "jdbc:phoenix:localhost:2181","table","RESULT")) 
     .save(); 
    } 
} 

Answer 3

からリード/ライトをする方法の例です。この、これは以下のように行うことができます。

import org.apache.phoenix.spark._ 
val sqlContext = spark.sqlContext 
val df1 = sqlContext.read.format("jdbc").options(Map("driver" -> "org.apache.phoenix.jdbc.PhoenixDriver","url" -> "jdbc:phoenix:zk4-habsem.lzmf1fzmprtezol2fr25obrdth.jx.internal.cloudapp.net,zk5-habsem.lzmf1fzmprtezol2fr25obrdth.jx.internal.cloudapp.net,zk1-habsem.lzmf1fzmprtezol2fr25obrdth.jx.internal.cloudapp.net:2181:/hbase-unsecure", "dbtable" -> "table_name")).load();