The main function
SparkConf SparkConf=new SparkConf (.) setAppName (" SparkMain ");
JavaSparkContext CTX=new JavaSparkContext (sparkConf);
BaseBean. InitConfig (CTX);
JavaRDD Datas=CTX. TextFile (input2, 1);
//get here is one of the json string
JavaPairRDD DataMap=datas. MapToPair (f - & gt; BaseUtil. ToDataTuple (f));
DataMap. SaveAsTextFile (output);


InitConfig BaseBean class method (CTX)
Public static Map RegionMap=new HashMap (a);
Public static void initConfig JavaSparkContext (CTX) throws the Exception {
JavaRDD RDDS=CTX. TextFile ("/data/conf/regionConf. TXT ");
List Regionss=RDDS. Collect ();
InitMap (regionMap regionss);
RegionMap. Put (" China & amp; City & amp; City ", "110100");
CTX. Broadcast (regionMap);
}

Public static void initMap (Map The map, List Ips) throws the Exception {
For (String IP: ips) {
String STR=new String (IP) getBytes (), "utf-8");
if(! BaseUtil. IsEmpty (STR)) {
String [] STRS=STR. The split ("=");
if(null !=STRS & amp; & STRS. Length==2) {
The map. The put (STRS [0], the STRS [1]).
If (STRS [0]. EndsWith (" area ")) {
Map. The put (STRS [0]. The substring (0, STRS [0]. The length () - 2) + "city", the STRS [1]).
} else if (STRS [0]. EndsWith (" state ")) {
Map. The put (STRS [0]. The substring (0, STRS [0]. The length () - 3) + "city", the STRS [1]).
} else if (STRS [0]. EndsWith (" union ")) {
Map. The put (STRS [0]. The substring (0, STRS [0]. The length () - 1) + "city", the STRS [1]).
} else if (STRS [0]. EndsWith (" & amp; City ")) {
Map. The put (STRS [0]. The substring (0, STRS [0]. The length () - 1), the STRS [1]).
} else if (STRS [0]. EndsWith (" city ")) {
Map. The put (STRS [0] + "city", the STRS [1]).
}
}
}

}
}


BaseUtil class toDataTuple method
Public static Tuple2 & lt; String, TopicBean> ToDataTuple (String STR) throws the Exception {
If (null==STR | | "" equals (STR) trim () | |" null ". EqualsIgnoreCase (STR)) {
Return new Tuple2 & lt; String, TopicBean> (" 0 ", null);
}
TopicBean TopicBean=GSON. FromJson (STR, TopicBean. Class);
String l1=topicBean. GetL1 ();
if(! BaseUtil. IsEmpty (l1)) {
If (l1. The contains (". ")) {
Return new Tuple2 & lt; String, TopicBean> (l1, topicBean);
} else {
String s15=topicBean. GetS15 ();
String s16=topicBean. GetS16 ();
String address=s15 + & amp; "" + s16 + & amp; "" + l1.
If (BaseBean getOtherCountry (address)!=0) {
TopicBean. SetL1 (String. The valueOf (BaseBean getOtherCountry (address)));
} else {
TopicBean. SetL1 (BaseUtil. ConverNull (BaseBean. RegionMap. Get (address)));
}
Return new Tuple2 & lt; String, TopicBean> (topicBean getL1 (), topicBean);
}
} else {
TopicBean. SetL1 (" 0 ");
Return new Tuple2 & lt; String, TopicBean> (" 0 ", topicBean);
}
}

Why my last the regionMap is always empty, not in front of the init

CodePudding user response:

CTX. Broadcast (regionMap); A return value, that is the cluster recognize radio out big variable, using the variable cluster computing get variables to discover your radio