最近在自學scala，在專案里找到了連接spark的代碼，想試試手用scala實作出來，卻發現有點行不通，java和scala的方法似乎并不是一樣的，我自己的scala也是半生不熟，求教各位大佬最后那個方法JavaRDD<String> u = gaaatem2.toJavaRDD().map()應該怎樣用scala實作，在下感激不盡
public class Gaaa_Wide_Yarn {
public static void main(String[] args) {
SparkConf sc = new SparkConf().setAppName("Gaaa_Wide_Yarn");
SparkContext jsc = new SparkContext(sc);
HiveContext hc = new HiveContext(jsc);
String day = args[0];
System.out.println("Gaaa_Wide_Yarn day:" + day);
String gaaaSql=" select username,groupid,starttime,stoptime,timelen,octets,sourceno,nasip,"
+ " nasport,frameip,downreason "
+ " from shjs_wlpt.interface_aaag_wide_day "
+ " where day = '"+day+"'";
String gaaaSql_1=" select m.cityname as area_name,m.citycrmid as area_code, username,groupid,starttime,stoptime,timelen,octets,sourceno,nasip,"
+ " nasport,frameip,downreason,'' as reserve1,'' as reserve2,'' as reserve3,'' as reserve4 "
+ " from gaaaSql t,usi_odso.itv_check_city m "
+ " where t.sourceno = m.cityid ";
System.out.println("gaaaSql_1:"+gaaaSql_1);
DataFrame gaaatem1=hc.sql(gaaaSql);
gaaatem1.registerTempTable("gaaaSql");
DataFrame gaaatem2=hc.sql(gaaaSql_1);
gaaatem2.registerTempTable("gaaaSql_1");
SimpleDateFormat df3 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
final String today = df3.format(new Date());
@SuppressWarnings("serial")
JavaRDD<String> u = gaaatem2.toJavaRDD().map(
new Function<Row, String>() {
public String call(Row row) {
String str = "";
str += today + "\001";//更新日期放首位，為后續加欄位方便
for (int i = 0; i < row.length(); i++) {
str += row.getString(i) + "\001";
}
return str;
}
});
u.repartition(20).saveAsTextFile(args[1]);
}
}
uj5u.com熱心網友回復：
Spark本身就是Scala寫的，你用Scala的API就好了嘛，為啥要用Java的？uj5u.com熱心網友回復：
這是專案里面寫好的java代碼啊，我想用scala實作
uj5u.com熱心網友回復：
真的哭了。。。
gaaatem2.rdd().map()
這里出來的rdd就是scala源生的rdd啊。。。
uj5u.com熱心網友回復：
意思是說只要直接gaaatem2.rdd就行了嗎，那里面那個對字串處理的方法應該寫在哪呢
uj5u.com熱心網友回復：
你好，我順手把你的java代碼 改成了scala版，看是不是符合你想要的
object Gaaa_Wide_Yarn {
def main(args: Array[String]): Unit = {
val sc = new SparkConf().setAppName("Gaaa_Wide_Yarn")
val jsc = new SparkContext(sc)
val hc = new HiveContext(jsc)
val day = args(0)
println("Gaaa_Wide_Yarn day:" + day)
val gaaaSql =" select username,groupid,starttime,stoptime,timelen,octets,sourceno,nasip,"
+ " nasport,frameip,downreason "
+ " from shjs_wlpt.interface_aaag_wide_day "
+ " where day = '"+day+"'";
val gaaaSql_1 = " select m.cityname as area_name,m.citycrmid as area_code, username,groupid,starttime,stoptime,timelen,octets,sourceno,nasip,"
+ " nasport,frameip,downreason,'' as reserve1,'' as reserve2,'' as reserve3,'' as reserve4 "
+ " from gaaaSql t,usi_odso.itv_check_city m "
+ " where t.sourceno = m.cityid ";
println("gaaaSql_1:"+gaaaSql_1)
val gaaatem1=hc.sql(gaaaSql)
gaaatem1.registerTempTable("gaaaSql")
val gaaatem2=hc.sql(gaaaSql_1);
gaaatem2.registerTempTable("gaaaSql_1");
val dfs3 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
val today = dfs3.format(new Date())
val u = gaaatem2.map(
new Function[Row,String](){
def call(row:Row):String = {
val str = ""
str += today + "\001"
for(i <- row.length){
str += row.getString(i)+"\001"
}
str
}
}
)
u.repartition(20).saveAsTextFile(args(1))
}
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/62927.html
標籤：Spark
上一篇：如何讓容災系統發揮最大作用