Shulou(Shulou.com)06/03 Report--

In the actual development process, we encounter a scenario that needs to achieve selective aggregation, that is, for the data corresponding to a certain key, the records that meet the conditions will be aggregated, and those that do not meet the conditions will not be aggregated.

When using spark to deal with this kind of computing scenario, you think of using the combineByKey operator to map the value in the input data to an ArrayBuffer with one element (equivalent to ArrayList in java in scala), and then overwrite the ArrayBuffer after aggregating the records that meet the aggregation conditions, and fill in the ArrayBuffer for the two records to be aggregated that do not meet the conditions. Finally, flatMap is called to split the elements in the ArrayBuffer.

For example, the following code implements selective aggregation according to time conditions when aggregating a field:

Val rdd1 = sc.textFile (dayDayDir) .union (sc.textFile (thisDayDir)) .map (line = > line.split ("\ |")) .filter (arr = > if (arr.length! = 14 | |! arr (3) .substring (0,8) .equals (lastDay) false else true) .map (arr = > (arr (0), arr)) .reduceByKey ((pure, after) = > reduceSession (pure, after)) .map (tup = > (tup._2 (13)) Tup._2) .combineByKey (x = > ArrayBuffer (x), (x: ArrayBuffer [Array [string]], y) = > combineMergeValue (x, y), (x: ArrayBuffer [ArrayString], y: ArrayBuffer [ArrayString]) = > combineMergeCombiners (x, y)) .flatMap (tup = > arrToStr (string)) def combineMergeValue (x: ArrayBuffer [ArrayString]] Y:Array [String]: ArrayBuffer [Array [string]] = {var outList = x.clone () var outarr = y.clone () var flag = true for (I

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.