Sp数据基础技术应用 1

广告检测的流量作弊识别

——Spark综合实战项目背景互联网虚假流量，是指通过特殊的方式，模仿人类浏览行为生成的访问流量。如通过设置程序，每分钟访问一次某网站的主页，这样的流量即属于虚假流量。浏览量的增加一般情况下可以促进销售量的增加。同等条件下，流量大的网站收取的广告费用更高，因此，部分网站受利益的驱使，会通过作弊方式产生虚假流量。虚假流量问题在数字营销行业中一直存在，给广告主带来了严重的损失。网站凭借虚假流量收取了更高的广告费用，却未促进真实浏览量和销售量的增加，直接损害了广告主的利益。本项目将通过Spark大数据技术实现广告检测的流量作弊识别，使读者可以更加熟悉Spark相关技术，并灵活应用相关技术解决相应的大数据问题。分析需求探索分析广告流量数据预处理数据并构建特征构建与评估分类模型常见的流量作弊方式流量作弊方式说明脚本刷量通过设定程序，使电脑按一定的规则访问目标网站控制肉鸡访问利用互联网上受病毒感染的电脑访问目标网站页面代码修改通过病毒感染或其他方式，在媒体网站插入隐藏代码，在其页面加载肉眼不可见的指向目标网站的小页面DNS劫持通过篡改DNS服务器上的数据，强制修改用户电脑的访问位置，使原本访问网站被动修改为目标网站需求分析需求分析虚假流量识别实现流程的步骤如下。对广告检测中获得的历史流量数据进行选择性抽取和数据划分。对第一步中形成的数据集进行数据探索分析，包括缺失值、冗余字段的基础探索和流量作弊的行为特征的业务探索。根据探索分析结果得出的清洗规则，对数据进行相应的预处理，并构建建模时需要的特征，形成建模样本数据。建立不同的虚假流量识别模型，并对模型进行评估及对比。保存效果较好的模型，模拟新数据产生，加载保存好的模型进行应用。分析需求探索分析广告流量数据预处理数据并构建特征构建与评估分类模型数据说明case_data_new.csv部分数据字段说明：字段名称说明rank记录序号dt相对日期，单位为天cookiecookie值ipIP地址，已经过脱敏idfaidfa值，可用于识别IOS用户imeiimei值，可用于识别Android用户androidandroid值，可用于识别Android用户openudidopenudid值，可用于识别IOS用户macmac值，可用于识别不同硬件设备timestamps时间戳camp项目IDcreativeid创意IDmobile_os设备OS版本信息，该值为原始值本任务选择本地模式对获取到数据进行基本的探索分析工作，根据探索分析得出的数据清洗规则，对数据进行进一步的处理。首先创建一个Spark工程，导入Spark相关的开发依赖包，创建一个Explore.scala类。在Explore.scala类的主函数中创建SparkSession对象，命名为Spark，并设置日志输出等级为“WARN”。基础探索数据importorg.apache.spark.sql.SparkSessionobjectExplore{defmain(args:Array[String]):Unit={valspark=SparkSession.builder().appName("Explore").master("local[*]").enableHiveSupport().getOrCreate()spark.sparkContext.setLogLevel("WARN")}}探索记录数读取本地case_data_new.csv文件的数据为DataFrame格式的数据，通过option设置文件首行为列名，使用count()方法计算数据记录数。结果如图所示，7天的流量数据，记录数共有1704154条。valrawData=spark.read.option("header","true").csv("E:\\data\\case_data_new.csv")println("原始数据集行数为："+rawData.count())探索日流量广告检测流量数据记录总共有7天，在样本数据中，dt字段记录了流量数据提取的相对天数。dt字段的值为1~7，1表示提取的7天流量数据的第一天数据。以此类推。对每天的数据流量数进行统计，查看是否有异常现象。使用groupBy()方法根据dt字段进行分组统计，查询7天中每天的访问量，并根据相对天数进行升序排序。如图所示，可以看出日流量差异不大，数据产生的环境相对稳定，不存在数据倾斜问题。rawData.groupBy("dt").count().selectExpr("dt","countasdayCount").sort("dt").show()分析数据类型如表所示，大部分的变量是字符类型，不适用于直接运用至一些需要数值类型的分类模型中。因此后续需要对字符类型的数据进行相应的编码或根据字符类型的数据字段构造新的特征列。变量类型变量名称字符型（character）cookie、ip、idfa、imei、android、openudid、mac、app_key_md5、app_name_md5、placementid、useragent、os_type数值型（numeric）rank、dt、timestamps、camp、creativeid、mobile_os、mobile_type、mediaidborn_time、label统计缺失数据本次任务对7天中所有的广告流量数据进行缺失值探索，统计出各个属性的数量情况。使用SparkSQL的DataFrameAPI操作接口中的na()方法对缺失值进行统计，结合drop()方法即可得到数据字段的缺失率。通过观察，creativeid字段大量值都为0，不符合正常情况，因此creativeid字段为0视为缺失值。由于creativeid字段不同与其他字段，所以需单独统计。为提供代码利用率，在主程序（main()方法）外构建自定义方法用于计算数据字段缺失率。统计缺失数据自定义方法统计缺失率：defMissingCount(data:DataFrame,columnName:String):Unit={varmissingRate=0.0if(columnName!="creativeid"){missingRate=1-(data.select(columnName).na.drop().count().toDouble/data.count())}else{missingRate=data.select(columnName).filter("creativeid==0").count()/data.count().toDouble}if(missingRate>0){println(columnName+"缺少值比率："+missingRate*100+"%")}}统计缺失数据在主程序内调用for循环对各个字段进行统计：结果如图所示，原数据中共有22个字段，存在缺失值的有12个字段。useragent字段的缺失率较少，剩余字段的缺失率很高，尤其是creativeid字段高达98.39%，且原始数据为数值型数据，无法进行插补，因此，后续将编写程序删除缺失率过高的数据字段。valcolumnName=rawData.columns.toListfor(i<-columnName){MissingCount(rawData,i)}冗余数据数据集成往往会导致数据冗余，而数据冗余往往会影响模型构建的速度和质量。冗余字段缺失率分析结果：结果如表所示，idfa、imei、android和openudid这4个字段的缺失率是偏高的，而且对这4个字段进行字段识别，发现均是用于识别手机系统类型的字段。因此，若后续构建特征时需要使用识别手机系统类型的字段，可以将这4个字段合并，以降低手机系统类型属性的缺失率，提取有效信息。字段名称缺失率备注idfa92.19%可用于识别IOS用户imei79.83%可用于识别Android用户android80.79%可用于识别Android用户openudid84.14%可用于识别IOS用户脚本刷新网页作弊使用脚本刷新网页作弊时，产生的数据记录中cookie与ip不变，且存在多条记录。因此在cookie和ip不变的情况下统计数据记录数，使用groupBy()方法根据ip和cookie字段进行分组统计。使用withcolumn()新增一列ip_cookie_count_precent，用于存放cookie和ip相同的数据记录数。valcookie_ip_distribute=rawData.groupBy("ip","cookie").count().withColumn("ip_cookie_count_precent",col("count")/rawData.count()*100).orderBy(desc("count"))cookie_ip_distribute.show(false)脚本刷新网页作弊结果如图所示，在7天的流量数据中，确实存在同一ip和cookie高频浏览广告的情况。脚本刷新网页作弊正常情况下极少会有人在7天内频繁点击某广告多达100次或以上，因此在ip和cookie不变的情况下，对广告的浏览频数超过100次的记录数进行简单统计。使用filter()方法，过滤出代码统计cookie和ip相同的流量记录数中count字段大于100的数据，并统计筛选后的数据记录数。结果如图所示，7天中同一个ip和cookie浏览广告超过100次的有104个，将所有的访问次数进行叠加，将占据大量的流量记录，如果不加以识别，那么将对广告主造成很大损失。valclick_gt_100=cookie_ip_distribute.filter("count>100").count()println("同ip、cookie出现超过100次以上的记录数："+click_gt_100)定期清除cookie，刷新网页作弊同一ip和cookie浏览数据很容易被进行识别，因此作弊者往往也会通过定期清除cookie，进而产生新的cookie，制造不同cookie的访问记录。该类的虚假流量得特征表现：ip不变，多条不同cookie记录。使用groupBy()方法根据ip字段进行分组统计，统计每个ip对应的不同cookie次数的分布情况，valip_distribute=rawData.groupBy("ip").agg(countDistinct("cookie")as"ip_count").groupBy("ip_count").agg(count("ip_count")as"ip_count_count",count("ip_count")/rawData.count()*100as"ip_count_count_precent").orderBy(desc("ip_count"))ip_distribute.show(false)定期清除cookie，刷新网页作弊结果显示存在同一个ip高频访问广告的情况，且占据着较高的数据比例，需要进行识别。ADSL重新拨号后刷新网页作弊利用ADSL重新拨号后刷新网页作弊，其特征是在某一时间段里，多条访问记录的ip来源于同个区域，因此ip的前两段或前3段相同。根据ip前两段进行分组统计，统计ip前两段相同的记录数分布情况，使用substring_index()方法根据“.”对ip地址进行分割。取出前两段，统计前两段ip相同的记录数，并根据记录数进行降序排序。valip_two=rawData.withColumn("ip_two",substring_index(col("ip"),".",2)).groupBy("ip_two").agg(count("ip_two")as"ip_two_count").orderBy(desc("ip_two_count"))ip_two.show(false)ADSL重新拨号后刷新网页作弊结果如图所示，ip前两段相同的流量记录数在10000以上的情况较多。ADSL重新拨号后刷新网页作弊根据ip前3段进行分组统计，统计ip前3段相同的记录数的分布情况。结果如图所示，ip前3段相同的流量记录上万的情况也较多，与ip前两段相同一样，如果对应的记录数过于庞大，则可以大致判定为虚假流量数据。valip_three=rawData.withColumn("ip_three",substring_index(col("ip"),".",3)).groupBy("ip_three").agg(count("ip_three")as"ip_three_count").orderBy(desc("ip_three_count"))ip_three.show(false)分析需求探索分析广告流量数据预处理数据并构建特征构建与评估分类模型删除缺失值字段在之前的数据探索分析中，由于缺失率较高的数据字段多为字符型的字段，所以无法对缺失值进行插补，为了减小缺失数据对模型产生的影响，删除缺失率过高的mac、creativeid、mobile_os、mobile_type、app_key_md5、app_name_md5、os_type字段。对于idfa、imei、android、openudid这4个数据含义相似的数据字段，由于后续构建特征时不确定是否需要使用到，所以先不进行处理。valdata_new=rawData.drop("mac").drop("creativeid").drop("mobile_os").drop("mobile_type").drop("app_key_md5").drop("app_name_md5").drop("os_type")删除缺失值字段将处理后的数据保存至Hive中。依次启动Hadoop集群和Hive元数据服务，并进入HiveCLI中创建数据库ad_traffic。将hive-site.xml配置文件复制至项目工程的resources目录中。在Spark程序中使用saveAsTable()方法中将处理后的保存至Hive的ad_traffic数据库中，表名为AdData，保存模式通过mode()方法设置为覆盖保存。data_new.write.mode("overwrite").saveAsTable("ad_traffic.AdData")删除缺失值字段待程序运行完成后，在HiveCLI中查看保存的AdData数据表的字段结构，结果如图所示。构建广告流量作弊识别特征对于分类模型，构建的特征对于标签列的识别需要存在一定的判别标准，之前已对不同作弊行为对应的虚假流量数据特征进行探索，根据探索结果，分别构建N、N1、N2、N3特征。5小时是根据广告点击周期的频率进行划分得出。特征构建方法说明N统计在5个小时内，原始数据集中，ip与cookie两个属性相同的记录出现次数ip和cookie不变的情况下，出现的记录次数指标：NN1统计在5个小时内，原始数据集中，同一个ip产生的cookie记录条数ip不变，对应cookie出现的记录次数指标：N1N2统计在5个小时内，原始数据集中，ip前两段相同的记录的出现次数ip前两段相同的次数指标：N2N3统计在5个小时内，原始数据集中，ip前3段相同的记录的出现次数ip前3段相同的次数指标：N3构建广告流量作弊识别特征将7天按5小时进行划分会得到多个区间，再对每个区间内部的数据进行特征构建，同时将每个区间的特征数据合并至同一个数据集中作为特征数据集。实现过程的大体思路框架如下：首先获取时间戳的最大值和最小值，以最大时间点和最小时间点作为界限，以18000秒（即5小时）作为分割点，构建时间区间。使用for循环对构建的时间区间进行遍历，获取对应区间内的数据，进行特征构建。构建广告流量作弊识别特征由于Scala语言中的for循环内部构建的变量无法传出for循环之外，所以无法在代码呈现中完成各个时间区间特征数据集的拼接。但可以在循环中将每个区间内构建的特征保存至Hive表中，当循环完成时，循环内所构建的特征数据集将保存至同一个Hive表中。由于数据量非常大，同时需要构建每个区间内的特征，这一系列的操作对硬件要求极大。若要对7天的广告流量数据进行处理会耗费大量的时间，所以取出前25个小时的广告流量数据作为构建特征前的数据集。划分时间区间range()方法可以在自定义的区间范围内，以规定的间隔对自定义的区间进行等分切割成不同的小区间。importorg.apache.spark.sql.SparkSessionimportorg.apache.spark.sql.functions.{col,max,min}importorg.apache.spark.sql.types.DataTypesobjectFeatures{defmain(args:Array[String]):Unit={valspark=SparkSession.builder().appName("Features").master("local[*]").enableHiveSupport().getOrCreate()spark.sparkContext.setLogLevel("WARN")valtimestamps="timestamps"valcookie="cookie"valip="ip"valN="N"valN1="N1"valN2="N2"valN3="N3"valranks="rank"划分时间区间区间分割点的部分结果如图所示。此处将以18000秒为切割间隔将数据切割为多个区间。valdata=spark.read.table("ad_traffic.AdData")valmax_min_timestamp=data.select(max(col(timestamps).cast(DataTypes.IntegerType))as"max_ts",min(col(timestamps).cast(DataTypes.IntegerType))as"min_ts").rdd.collect()valmax_ts=max_min_timestamp(0).getInt(0)valmin_ts=max_min_timestamp(0).getInt(1)valtimes=List.range(min_ts,max_ts,18000)println("时间分割点："+times)}}构建特征得到时间划分的区间列表后，需要根据区间列表取出前25小时的数据。使用for循环，再通过filter()方法筛选出timestamps字段在区间范围内的数据。筛选得到前25个小时内的数据后，以5小时的区间对数据进行特征构建，构建特征N、N1、N2、N3。在得到4个特征数据集后，将这些数据集以ranks字段进行合并得到含rangs和4个特征的完整特征数据集。将数据集以append的方式写入hive表中，这时hive表中就将会存在5个小时的特征数据集，之后进行下一次的循环，特征4个独立的特征数据集将会被重新赋值、合并，添加至Hive表中。在Features.scala类添加代码进行特征构建与数据保存。构建特征for(i<-0to4){valdata_sub=data.filter("timestamps>="+times(i)+"andtimestamps<"+times(i+1))valdata_N_sub=data_sub.groupBy(cookie,ip).agg(count(ip)asN).join(data_sub,Seq(cookie,ip),"inner").select(ranks,N)valdata_N1_sub=data_sub.groupBy(ip).agg(countDistinct(cookie)asN1).join(data_sub,Seq(ip),"inner").select(ranks,N1)valdata_ip_two=data_sub.withColumn("ip_two",substring_index(col(ip),".",2))valdata_ip_three=data_sub.withColumn("ip_three",substring_index(col(ip),".",3))valdata_N2_sub=data_ip_two.groupBy("ip_two").agg(count("ip_two")as"N2").join(data_ip_two,Seq("ip_two"),"inner").select(ranks,N2)valdata_N3_sub=data_ip_three.groupBy("ip_three").agg(count("ip_three")as"N3").join(data_ip_three,Seq("ip_three"),"inner").select(ranks,N3)valdata_model_N=data_N_sub.join(data_N1_sub,ranks).join(data_N2_sub,ranks).join(data_N3_sub,ranks)data_model_N.write.mode("append").saveAsTable("ad_traffic.TimeFeatures")}构建特征待程序运行完成后，在HiveCLI中查看保存的TimeFeatures数据表的字段结构，并查询数据表前10行数据示例。TimeFeatures表中只存在5个数据字段，不包含label字段，label字段存在于完整数据集中。因此需要将TimeFeatures和AdData表进行合并连接。构建特征完成特征构建后，读取TimeFeatures表和AdData表的数据并根据ranks字段进行连接，选取4个特征字段、dt字段和label字段，并保存至FeaturesData表中。待程序运行完成后，在HiveCLI中查看保存的FeaturesData数据表的字段结构，并查询数据表前10行数据示例，如图所示。valFeaturesData=spark.read.table("ad_traffic.TimeFeatures").join(data,ranks).select(col(ranks),col("dt"),col("N"),col("N1"),col("N2"),col("N3"),col("label").cast("double"))FeaturesData.write.mode("overwrite").saveAsTable("ad_traffic.FeaturesData")特征标准化如果特征之间的值存在很大的差异，那么可能会导致某一特征对模型的预测结果有着更大且不合理的影响，因此需要对特征数据进行标准化处理。由于特征数据之间的差值较大，因此将使用最小值——最大值归一化进行处理：在项目工程中新建一个Scaler.scala类。将需要进行归一化的字段合并至同一个向量中。使用MinMaxScaler()方法进行处理。特征标准化进行数据归一化后，对数据集进行划分，使用randomSplite()方法将数据按7:3比例进行划分。分别保存为modelData模型构建数据和testData模型加载数据，modelData用于后续的模型构建与评估，testData则用于模拟真实的模型应用的阶段。结果如图所示，处理后的数据集总数据量为702135，按照7:3的比例进行划分后的模型构建数据和模型加载数据的数据量分别为491698和210437。valArray(modelData,loadData)=dataScaler.randomSplit(Array(0.7,0.3))modelData.write.mode("overwrite").saveAsTable("ad_traffic.ModelData")loadData.write.mode("overwrite").saveAsTable("ad_traffic.LoadData")println("数据集总数据量："+dataScaler.count())println("模型构建数据量："+modelData.count())println("模型加载数据量："+loadData.count())分析需求探索分析广告流量数据预处理数据并构建特征构建与评估分类模型构建与评估逻辑回归算法模型通过观察label标签可以看出，广告流量作弊识别为经典的二分类问题，即该广告访问记录是否为作弊访问记录。逻辑回归是二分类问题上的一个经典模型，而且逻辑回归原理简单，对于二分类的预测准确率也较高。构建逻辑回归模型，并评估起效果。在模型构建与评估中，编写的Spark程序不再以本地模式运行，而是对Spark程序进行编译打包，使用集群模式将程序提交至集群中运行。因此在IDEA中的SparkSession的实例化和部分参数的设置将会进行调整。构建逻辑回归模型本任务使用LogisticRegression()方法设置相关正则化系数和最大迭代次数等相关参数。经过参数调优后发现当以将最大迭代次数设置为15次，正则化系数设为0.03，其余相关参数以默认值进行构建时，模型效果较好。使用randomSplit()方法先将数据集按7:3的比例随机划分为训练集和测试集，调用LogisticRegression算法进行模型构建。逻辑回归算法所使用的数据格式为LabeledPoint格式，LogisticRegression算法会自动搜寻列名为features为特征字段，列名label为标签字段。因此构建模型时无需进行特征字段和标签字段的参数设置。评估逻辑回归模型SparkMLlibAPI对于模型提供多种的模型评估标准，因此选择最为直观的accuracy进行评估。使用MulticlassClassificationEvaluator对象构建模型评估器，设置标签字段名称和预测标签字段名称，并设置评估标准为accuracy，对测试数据的预测结果进行准确率计算。当对模型完成构建、训练、评估一系列操作后，可以将效果优良的模型进行保存。若后续需要对未知类别的广告流量数据进行作弊识别时，可直接加载保存好的模型并进行应用，模型的保存方法。valpre=model.transform(test)pre.select("label","prediction").show()valevaluator=newMulticlassClassificationEvaluator().setLabelCol("label").setPredictionCol("prediction").setMetricName("accuracy")println("LogisticModelAccuracy："+evaluator.evaluate(pre))model.write.overwrite().save(output)评估逻辑回归模型将项目工程编译打包，命名为model.jar。将JAR包上传至Linux的/opt/data目录下，使用spark-submit提交程序至集群中运行。逻辑回归模型预测的准确率较高，为85.8%，对于大部分的作弊访问记录都可以进行识别。spark-submit--masterspark://master:7077\--classLogistic/opt/data/model.jarad_traffic.ModelData/tipdm/data/model/LogisticSpark程序的运行模式设置为Spark的独立集群模式。设置输入数据为ad_traffic.ModelData表，设置模型的保存路径为HDFS的/tipdm/data/model/Logistic目录。importorg.apache.spark.ml.classification.RandomForestClassifierimportorg.apache.spark.sql.SparkSessionobjectRandomForest{defmain(args:Array[String]):Unit={valspark=SparkSession.builder().appName("RandomForest").master("spark://master:7077").enableHiveSupport().getOrCreate()spark.sparkContext.setLogLevel("WARN")valinputTable=args(0)valoutput=args(1)构建随机森林模型完成相应的实例化、数据读取及数据分割后，使用RandomForestClassifer()方法设置特征字段名为features，标签字段名为label，valModelData=spark.read.table(inputTable)valArray(train,test)=ModelData.randomSplit(Array(0.7,0.3))valrfModel=newRandomForestClassifier().setFeaturesCol("features").setLabelCol("label").setNumTrees(5).fit(train)}}评估随机森林模型构建随机森林分类模型训练数据后，使用该模型对测试数据进行预测。由于逻辑回归算法和随机森林算法构建的模型需要进行比较，所以使用同一种评估方法评估两个模型之间的优劣。模型评估依旧使用MulticlassClassificationEvaluator对象评估模型的准确度。valpre=rfModel.transform(test)pre.select("label","prediction").show()valevaluator=newMulticlassClassificationEvaluator().setLabelCol("label").setPredictionCol("prediction").setMetricName("accuracy")println("RandomForestModelAccuracy："+evaluator.evaluate(pre))rfModel.write.overwrite().save(output)评估随机森林模型将项目工程重新编译打包，命名为model.jar（覆盖打包）。将JAR包上传至Linux的/opt/data目录下。使用spark-submit提交程序至集群中运行。spark-submit--masterspark://master:7077\--classRandomForest/opt/data/model.jarad_traffic.ModelData/tipdm/data/model/RandomForest设置输入数据为ad_traffic.ModelData表，设置模型的保存路径为HDFS的/tipdm/data/model/RandomForest目录。评估随机森林模型随机森林模型的准确率约为91.1%，对比逻辑回归模型，准确率提高了近5%左右。因为逻辑回归模型的实现比较简单，本质是在线性回归模型上加入sigmoid函数，但在进行模型预测时很容易出现梯度消失的情况，而随机森林在数据足够的情况下分类器的准确率更高。模型加载加载任务之前保存的逻辑回归模型和随机森林模型，逻辑回归模型使用LogisticRegressionModel的load()方法。随机森林模型使用RandomForestClassificationModel的load()方法对模型进行加载。读取预处理数据并建构任务中分割出的LoadData数据集，模拟新产生的数据。使用加载好的模型对新数据进行预测。importorg.apache.spark.ml.classification.{LogisticRegressionMode