hsg77
commented
5 years ago `public class geowaveSpark {
public String main(String[] args)
throws Exception
{
String rbc="";
if (args==null) throw new IOException("args is null");
if (args.length!=3) throw new IOException("args have three params storeName fdName sparkMaster");
//
//String master = "spark://mycluster:7077";
Configuration hbaseConf =geowaveUtil.GetHBaseConfiguration(); // HBaseConfiguration.create();
//hbaseConf.set("hbase.zookeeper.quorum", app.zookeeper);
//hbaseConf.set("fs.defaultFS", "hdfs://mycluster/");
//System.setProperty("HADOOP_USER_NAME", app.zookeeperUser);
//
String storeName=args[0]; //dltb //cbdk
String fdName=args[1]; //dlbm //DKLB
String sparkMaster=args[2]; //local or local[*] or spark://node111:7077 or yarn
int minSplits = -1;
int maxSplits = -1;
//
SparkConf sparkConf=null;
//JavaSparkContext context =null;
SparkSession sparkSession=null;
DataStorePluginOptions inputStoreOptions = null;
try {
//env set
System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
System.setProperty("spark.kryo.registrator", "mil.nga.giat.geowave.analytic.spark.GeoWaveRegistrator");
// Attempt to load input store.
inputStoreOptions= geowaveUtil.getNewDataStorePluginOptions_hbase(storeName);
//============
SparkConf conf = GeoWaveSparkConf.getDefaultConfig()
.set("spark.yarn.jars","hdfs://mycluster:8020/spark/spark-libs.jar")
.set("spark.jars","hdfs://mycluster:8020/spark/spark-libs.jar")
//
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.set("spark.kryo.registrator", "mil.nga.giat.geowave.analytic.spark.GeoWaveRegistrator")
//
.set("spark.dynamicAllocation.enabled","false")
//
.set("spark.driver.extraClassPath","/usr/cwgis/app/spark/jars/lib/*")
.set("spark.executor.extraClassPath","/usr/cwgis/app/spark/jars/lib/*")
//
.set("spark.driver.memory","8g")
.set("spark.executor.memory","8g")
.set("spark.locality.wait","0")
.set("yarn.resourcemanager.address", "mycluster:8032")// resourcemanager
.set("yarn.resourcemanager.scheduler.address", "mycluster:8034")//
.set("spark.sql.broadcastTimeout", "1200")
.set("spark.sql.crossJoin.enabled", "true")
//
//.setMaster("local") //OK
//.setMaster("local[*]") //OK
.setMaster(sparkMaster) //???
.setAppName("geowaveSpark");
sparkSession = GeoWaveSparkConf.createDefaultSession(conf);
//============
//sparkConf.setJars(JavaSparkContext.jarOfClass(cls));
System.out.println("sparkMaster="+sparkMaster);
//
RDDOptions rddOptions=new RDDOptions();
rddOptions.setQueryOptions(null);
rddOptions.setQuery(null);
rddOptions.setMaxSplits(maxSplits);
rddOptions.setMinSplits(minSplits);
//-------
GeoWaveRDD gwRdd = GeoWaveRDDLoader.loadRDD(
sparkSession.sparkContext(),
inputStoreOptions,
rddOptions);
System.out.println("loaded :GeoWaveRDDLoader.loadRDD");
JavaPairRDD<GeoWaveInputKey, SimpleFeature> javaRdd=gwRdd.getRawRDD();
System.out.println("loaded:getRawRDD");
System.out.println("RDD feature count：" + javaRdd.count() +" of layerName：" + storeName + ""); //report error:kryo readObject
System.out.println("executing javaRdd.mapToPair");
JavaPairRDD<String, Double> mapToPair1 = javaRdd.mapToPair((feat)->{
Object o1 = feat._2.getAttribute(fdName);
if (o1==null) o1="null";
Double outValue=0.0;
Geometry geo=(Geometry)feat._2.getDefaultGeometry();
if(geo!=null)
{
outValue=geo.getArea();
if(outValue==0.0)
{
outValue=geo.getLength();
}
}
else
{
outValue=0.0;
}
return new Tuple2<String, Double>(new String(o1.toString()), outValue);
});
JavaPairRDD<String, Double> reduceByKey = mapToPair1.reduceByKey((x,y)->x+y);
Map<String, Double> mapToPair2 = reduceByKey.collectAsMap();
//================
rbc=this.getResult(mapToPair2,fdName);
System.out.println(rbc);
//================
javaRdd=null;
mapToPair1=null;
reduceByKey=null;
//
}
catch (Exception ex){
ex.printStackTrace();
}
finally {
//context.stop();
//context.close();
//context=null;
if(sparkSession!=null)
{
sparkSession.stop();
sparkSession.close();
sparkSession=null;
}
}
return rbc;
}
private String getResult(Map<String, Double> mapToPair2,String fdName)
{
Object objVal=null;
double area=0.0;
int dtRowsCount=0;
StringBuilder jsonBuilder = new StringBuilder();
//
SortedMap<String,Double> sortMap=new TreeMap<String,Double>(mapToPair2);
Set<String> keyset=sortMap.keySet();
for (String key : keyset) {
area=sortMap.get(key);
//
StringBuilder fdsBuilder=new StringBuilder();
fdsBuilder.append("{");
fdsBuilder.append("\""+fdName+"\":\"" + key + "\"");
fdsBuilder.append(",\"mj\":\"" + area + "\"");
fdsBuilder.append("}");
//
if(jsonBuilder.toString().length()<=0) {
jsonBuilder.append(fdsBuilder.toString());
}
else {
jsonBuilder.append(","+fdsBuilder.toString());
}
fdsBuilder=null;
dtRowsCount+=1;
}
return jsonBuilder.toString();
}}`
rfecher
Question about geowave 0.9.8 execution error: java.io.EOFException At org.apache.spark.serializer.KryoDeserializationStream.readObject
Master is local or local [*] execution is OK。 When Master is spark://mycluster:7077, execution will report KryoDeserializationStream.readObject error message.