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

How to use spark-core to achieve breadth-first search, for this problem, this article introduces the corresponding analysis and solutions in detail, hoping to help more partners who want to solve this problem to find a more simple and easy way.

Requirement description

The data source is a batch of network log data. Each piece of data has two fields, srcip and dstip, separated by commas. The requirement of the problem is to retrieve all the communication paths between the two ip given a srcip and dstip at a given search depth. This problem is a practical requirement in web log processing, which has been implemented in stand-alone programs before, but all ip pairs need to be loaded into memory. Considering that if the amount of data is too large, the memory of a single node may not be able to support such an operation, but if the ip is not fully loaded into memory and depth-first traversal is used, the search process will be very slow. Recently, I have been learning the spark framework, and I have just come into contact with RDD, which uses RDD to solve this problem. Here is the scala code

Package com.pxu.spark.coreimport org.apache.spark. {HashPartitioner, SparkConf SparkContext} / * pxu * 2021-01-29 16:57 * / object FindIpRel {def main (args: Array [String]): Unit = {val srcIp = args (0) / / Source ip val dstIp = args (1) / / destination ip val depth = args (2). ToInt / / search depth val resPath = args (3) / / output location of search results val conf = new SparkConf (). SetAppName ("findIpRel ") val sc = new SparkContext (conf) / * build the original rdd from the data source The data form of each row is a hdfs://master:9000/submitTest/input/ipconn/srcdst.csv b * / val ori = sc.textFile ("hdfs://master:9000/submitTest/input/ipconn/srcdst.csv") / *. The original Rdd is converted in tuple form, and now the data form of each row is (Aline b) * in addition, the data is deduplicated. It also shows that the data in the RDD is partitioned using the hash divider * for subsequent join operations, do some optimizations * / val base = ori.map (a = > {val tmpArr = a.split (",") (tmpArr (0), tmpArr (1)}). Distinct (). PartitionBy (new HashPartitioner (10)) / * this is a RDD used to save the results Each line is in the form of (dstIp,List (ip on path)) * during the search process, when the search result is found, it will be merged into res * / var res = sc.makeRDD [(String, list [string]) (List ()) / * this is an iterative RDD. Its initialization is that the first element an of the tuple is filtered from the baseRdd as the parameter SrcIp. * and then convert it to the format of (bjingle list (a)), where b always represents the tail ip on the current search path The rest of the list represents the search for * other ip * / var iteration = base.filter (_. _ 1.equals (srcIp)) .map (a = > (a.class2List (a.room1) for (I! a._2._1.contains (a.class2.room2)) .map (a = > (a.class2) A._2._1:+a._1) / * filter out the paths that have been successfully searched in tmp The criteria for successful search are (c, (List (ip on path) C is equivalent to dstIp * / val success = tmp.filter (a = > a._1.equals (dstIp)) / * merge successfully searched data into res * / res = res.union (success) / * Update iteration * / iteration = tmp.subtract (success)} / * * * merge the successfully searched path into res * / res.union (iteration.filter (a = > a._1.equals (dstIp) / * * perform a conversion operation Convert the elements in res from (c, (List (ip on path)) format to List (all ip on path) * / val finalResult = res.map (a = > a.room2: + a.room1) finalResult.saveAsTextFile (resPath)}} the answer to the question on how to use spark-core to achieve breadth-first search is shared here. I hope the above content can be of some help to you, if you still have a lot of questions unsolved. You can follow the industry information channel for more related knowledge.

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.