pom.xml

org.apache.spark

spark-sql_2.11

2.2.1

org.apache.phoenix

phoenix-core

4.13.1-HBase-1.2

org.apache.phoenix

phoenix-spark

4.13.1-HBase-1.2

joda-time

joda-time

2.9.9

调用示例

public class Test {

/**

* phoenix jdbc config

*/

private static final String DB_PHOENIX_DRIVER = "org.apache.phoenix.jdbc.PhoenixDriver";

private static final String DB_PHOENIX_URL = "jdbc:phoenix:hadoop101,hadoop102,hadoop103";

private static final String DB_PHOENIX_USER = "";

private static final String DB_PHOENIX_PASS = "";

private static final String DB_PHOENIX_FETCHSIZE = "10000";

/**

* 加载数据查询SQL

*/

private static final String SQL_QUERY = "(SELECT date,member_id FROM events WHERE time>='%s' AND time rows = sparkSession

.read()

.jdbc(DB_PHOENIX_URL, query, connProp)

.filter("member_id != -1")

.javaRDD()

.mapToPair(r -> new Tuple2<>(

r.getString(0)

, r.getLong(1)

))

.distinct()

.groupByKey()

.map(r -> {

StringBuffer sb = new StringBuffer();

r._2.forEach(v -> {

sb.append(v);

});

return RowFactory.create(r._1, sb.toString());

});

// schema

Listfields = new ArrayList<>();

fields.add(DataTypes.createStructField("date", DataTypes.StringType, false));

fields.add(DataTypes.createStructField("dist_mem", DataTypes.StringType, true));

StructType schema = DataTypes.createStructType(fields);

// 写入

String insertTable = "test_string";

sparkSession

.createDataFrame(rows, schema)

.write()

.format("org.apache.phoenix.spark")

.mode(SaveMode.Overwrite)

.option("table", insertTable)

.option("zkUrl", DB_PHOENIX_URL)

.save();

} catch (Exception e) {

e.printStackTrace();

}

}

}

调用：

public class App {

@Test

public void testJob() {

String[] args = new String[]{"2017-06-01", "2017-07-01"};

Test.main(args);

}

}