[指定分隔符进行字符串分割 split与r...]的搜索结果

...我们，我们将第一时间进行核实并删除相应内容。 Spark Streaming电商广告点击综合案例 需求分析和技术架构 广告点击系统实时分析 广告来自于广告或者移动App等，广告需要设定在具体的广告位，当用户点击广告的时候，一般都会通过ajax或Socket往后台发送日志数据，在这里我们是要做基于SparkStreaming做实时在线统计。那么数据就需要放进消息系统（Kafka）中，我们的Spark Streaming应用程序就会去Kafka中Pull数据过来进行计算和消费，并把计算后的数据放入到持久化系统中（MySQL） 广告点击系统实时分析的意义：因为可以在线实时的看见广告的投放效果，就为广告的更大规模的投入和调整打下了坚实的基础，从而为公司带来最大化的经济回报。 核心需求： 1、实时黑名单动态过滤出有效的用户广告点击行为：因为黑名单用户可能随时出现，所以需要动态更新； 2、在线计算广告点击流量； 3、Top3热门广告； 4、每个广告流量趋势； 5、广告点击用户的区域分布分析 6、最近一分钟的广告点击量； 7、整个广告点击Spark Streaming处理程序724小时运行； 数据格式： 时间、用户、广告、城市等 技术细节： 在线计算用户点击的次数分析，屏蔽IP等； 使用updateStateByKey或者mapWithState进行不同地区广告点击排名的计算； Spark Streaming+Spark SQL+Spark Core等综合分析数据； 使用Window类型的操作； 高可用和性能调优等等； 流量趋势，一般会结合DB等； Spark Core / /package com.tom.spark.SparkApps.sparkstreaming;import java.util.Date;import java.util.HashMap;import java.util.Map;import java.util.Properties;import java.util.Random;import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig;/ 数据生成代码，Kafka Producer产生数据/public class MockAdClickedStat {/ @param args/public static void main(String[] args) {final Random random = new Random();final String[] provinces = new String[]{"Guangdong", "Zhejiang", "Jiangsu", "Fujian"};final Map<String, String[]> cities = new HashMap<String, String[]>();cities.put("Guangdong", new String[]{"Guangzhou", "Shenzhen", "Dongguan"});cities.put("Zhejiang", new String[]{"Hangzhou", "Wenzhou", "Ningbo"});cities.put("Jiangsu", new String[]{"Nanjing", "Suzhou", "Wuxi"});cities.put("Fujian", new String[]{"Fuzhou", "Xiamen", "Sanming"});final String[] ips = new String[] {"192.168.112.240","192.168.112.239","192.168.112.245","192.168.112.246","192.168.112.247","192.168.112.248","192.168.112.249","192.168.112.250","192.168.112.251","192.168.112.252","192.168.112.253","192.168.112.254",};/ Kafka相关的基本配置信息/Properties kafkaConf = new Properties();kafkaConf.put("serializer.class", "kafka.serializer.StringEncoder");kafkaConf.put("metadeta.broker.list", "Master:9092,Worker1:9092,Worker2:9092");ProducerConfig producerConfig = new ProducerConfig(kafkaConf);final Producer<Integer, String> producer = new Producer<Integer, String>(producerConfig);new Thread(new Runnable() {public void run() {while(true) {//在线处理广告点击流的基本数据格式：timestamp、ip、userID、adID、province、cityLong timestamp = new Date().getTime();String ip = ips[random.nextInt(12)]; //可以采用网络上免费提供的ip库int userID = random.nextInt(10000);int adID = random.nextInt(100);String province = provinces[random.nextInt(4)];String city = cities.get(province)[random.nextInt(3)];String clickedAd = timestamp + "\t" + ip + "\t" + userID + "\t" + adID + "\t" + province + "\t" + city;producer.send(new KeyedMessage<Integer, String>("AdClicked", clickedAd));try {Thread.sleep(50);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }} }).start();} } package com.tom.spark.SparkApps.sparkstreaming;import java.sql.Connection;import java.sql.DriverManager;import java.sql.PreparedStatement;import java.sql.ResultSet;import java.sql.SQLException;import java.util.ArrayList;import java.util.Arrays;import java.util.HashMap;import java.util.HashSet;import java.util.Iterator;import java.util.List;import java.util.Map;import java.util.Set;import java.util.concurrent.LinkedBlockingQueue;import kafka.serializer.StringDecoder;import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.Function;import org.apache.spark.api.java.function.Function2;import org.apache.spark.api.java.function.PairFunction;import org.apache.spark.api.java.function.VoidFunction;import org.apache.spark.sql.DataFrame;import org.apache.spark.sql.Row;import org.apache.spark.sql.RowFactory;import org.apache.spark.sql.hive.HiveContext;import org.apache.spark.sql.types.DataTypes;import org.apache.spark.sql.types.StructType;import org.apache.spark.streaming.Durations;import org.apache.spark.streaming.api.java.JavaDStream;import org.apache.spark.streaming.api.java.JavaPairDStream;import org.apache.spark.streaming.api.java.JavaPairInputDStream;import org.apache.spark.streaming.api.java.JavaStreamingContext;import org.apache.spark.streaming.api.java.JavaStreamingContextFactory;import org.apache.spark.streaming.kafka.KafkaUtils;import com.google.common.base.Optional;import scala.Tuple2;/ 数据处理，Kafka消费者/public class AdClickedStreamingStats {/ @param args/public static void main(String[] args) {// TODO Auto-generated method stub//好处：1、checkpoint 2、工厂final SparkConf conf = new SparkConf().setAppName("SparkStreamingOnKafkaDirect").setMaster("hdfs://Master:7077/");final String checkpointDirectory = "hdfs://Master:9000/library/SparkStreaming/CheckPoint_Data";JavaStreamingContextFactory factory = new JavaStreamingContextFactory() {public JavaStreamingContext create() {// TODO Auto-generated method stubreturn createContext(checkpointDirectory, conf);} };/ 可以从失败中恢复Driver，不过还需要指定Driver这个进程运行在Cluster，并且在提交应用程序的时候制定--supervise;/JavaStreamingContext javassc = JavaStreamingContext.getOrCreate(checkpointDirectory, factory);/ 第三步：创建Spark Streaming输入数据来源input Stream: 1、数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等 2、在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口的数据 (当然该端口服务首先必须存在），并且在后续会根据业务需要不断有数据产生（当然对于Spark Streaming 应用程序的运行而言，有无数据其处理流程都是一样的） 3、如果经常在每间隔5秒钟没有数据的话不断启动空的Job其实会造成调度资源的浪费，因为并没有数据需要发生计算；所以 实际的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；///创建Kafka元数据来让Spark Streaming这个Kafka Consumer利用Map<String, String> kafkaParameters = new HashMap<String, String>();kafkaParameters.put("metadata.broker.list", "Master:9092,Worker1:9092,Worker2:9092");Set<String> topics = new HashSet<String>();topics.add("SparkStreamingDirected");JavaPairInputDStream<String, String> adClickedStreaming = KafkaUtils.createDirectStream(javassc, String.class, String.class, StringDecoder.class, StringDecoder.class,kafkaParameters, topics);/因为要对黑名单进行过滤，而数据是在RDD中的，所以必然使用transform这个函数； 但是在这里我们必须使用transformToPair，原因是读取进来的Kafka的数据是Pair<String,String>类型, 另一个原因是过滤后的数据要进行进一步处理，所以必须是读进的Kafka数据的原始类型 在此再次说明，每个Batch Duration中实际上讲输入的数据就是被一个且仅被一个RDD封装的，你可以有多个 InputDStream，但其实在产生job的时候，这些不同的InputDStream在Batch Duration中就相当于Spark基于HDFS 数据操作的不同文件来源而已罢了。/JavaPairDStream<String, String> filteredadClickedStreaming = adClickedStreaming.transformToPair(new Function<JavaPairRDD<String,String>, JavaPairRDD<String,String>>() {public JavaPairRDD<String, String> call(JavaPairRDD<String, String> rdd) throws Exception {/ 在线黑名单过滤思路步骤： 1、从数据库中获取黑名单转换成RDD，即新的RDD实例封装黑名单数据； 2、然后把代表黑名单的RDD的实例和Batch Duration产生的RDD进行Join操作， 准确的说是进行leftOuterJoin操作，也就是说使用Batch Duration产生的RDD和代表黑名单的RDD实例进行 leftOuterJoin操作，如果两者都有内容的话，就会是true，否则的话就是false 我们要留下的是leftOuterJoin结果为false； /final List<String> blackListNames = new ArrayList<String>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doQuery("SELECT FROM blacklisttable", null, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {while(result.next()){blackListNames.add(result.getString(1));} }});List<Tuple2<String, Boolean>> blackListTuple = new ArrayList<Tuple2<String,Boolean>>();for(String name : blackListNames) {blackListTuple.add(new Tuple2<String, Boolean>(name, true));}List<Tuple2<String, Boolean>> blacklistFromListDB = blackListTuple; //数据来自于查询的黑名单表并且映射成为<String, Boolean>JavaSparkContext jsc = new JavaSparkContext(rdd.context());/ 黑名单的表中只有userID，但是如果要进行join操作的话就必须是Key-Value，所以在这里我们需要 基于数据表中的数据产生Key-Value类型的数据集合/JavaPairRDD<String, Boolean> blackListRDD = jsc.parallelizePairs(blacklistFromListDB);/ 进行操作的时候肯定是基于userID进行join，所以必须把传入的rdd进行mapToPair操作转化成为符合格式的RDD/JavaPairRDD<String, Tuple2<String, String>> rdd2Pair = rdd.mapToPair(new PairFunction<Tuple2<String,String>, String, Tuple2<String, String>>() {public Tuple2<String, Tuple2<String, String>> call(Tuple2<String, String> t) throws Exception {// TODO Auto-generated method stubString userID = t._2.split("\t")[2];return new Tuple2<String, Tuple2<String,String>>(userID, t);} });JavaPairRDD<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> joined = rdd2Pair.leftOuterJoin(blackListRDD);JavaPairRDD<String, String> result = joined.filter(new Function<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, Boolean>() {public Boolean call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> tuple)throws Exception {// TODO Auto-generated method stubOptional<Boolean> optional = tuple._2._2;if(optional.isPresent() && optional.get()){return false;} else {return true;} }}).mapToPair(new PairFunction<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, String, String>() {public Tuple2<String, String> call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> t)throws Exception {// TODO Auto-generated method stubreturn t._2._1;} });return result;} });//广告点击的基本数据格式：timestamp、ip、userID、adID、province、cityJavaPairDStream<String, Long> pairs = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t) throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} });/ 第4.3步：在单词实例计数为1基础上，统计每个单词在文件中出现的总次数/JavaPairDStream<String, Long> adClickedUsers= pairs.reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long i1, Long i2) throws Exception{return i1 + i2;} });/判断有效的点击，复杂化的采用机器学习训练模型进行在线过滤 简单的根据ip判断1天不超过100次；也可以通过一个batch duration的点击次数判断是否非法广告点击，通过一个batch来判断是不完整的，还需要一天的数据也可以每一个小时来判断。/JavaPairDStream<String, Long> filterClickedBatch = adClickedUsers.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {if (1 < v1._2){//更新一些黑名单的数据库表return false;} else { return true;} }});//filterClickedBatch.print();//写入数据库filterClickedBatch.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：userID,adID,clickedCount,time//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<UserAdClicked> userAdClickedList = new ArrayList<UserAdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");UserAdClicked userClicked = new UserAdClicked();userClicked.setTimestamp(splited[0]);userClicked.setIp(splited[1]);userClicked.setUserID(splited[2]);userClicked.setAdID(splited[3]);userClicked.setProvince(splited[4]);userClicked.setCity(splited[5]);userAdClickedList.add(userClicked);}final List<UserAdClicked> inserting = new ArrayList<UserAdClicked>();final List<UserAdClicked> updating = new ArrayList<UserAdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final UserAdClicked clicked : userAdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclicked WHERE"+ " timestamp =? AND userID = ? AND adID = ?",new Object[]{clicked.getTimestamp(), clicked.getUserID(),clicked.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(UserAdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getIp(),insertRecord.getUserID(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclicked VALUES(?, ?, ?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(UserAdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getTimestamp(),updateRecord.getIp(),updateRecord.getUserID(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity(),updateRecord.getClickedCount() + 1});}jdbcWrapper.doBatch("UPDATE adclicked SET clickedCount = ? WHERE"+ " timestamp =? AND ip = ? AND userID = ? AND adID = ? "+ "AND province = ? AND city = ?", updateParametersList);} });return null;} });//再次过滤，从数据库中读取数据过滤黑名单JavaPairDStream<String, Long> blackListBasedOnHistory = filterClickedBatch.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {//广告点击的基本数据格式：timestamp,ip,userID,adID,province,cityString[] splited = v1._1.split("\t"); //提取key值String date =splited[0];String userID =splited[2];String adID =splited[3];//查询一下数据库同一个用户同一个广告id点击量超过50次列入黑名单//接下来 根据date、userID、adID条件去查询用户点击广告的数据表，获得总的点击次数//这个时候基于点击次数判断是否属于黑名单点击int clickedCountTotalToday = 81 ;if (clickedCountTotalToday > 50) {return true;}else {return false ;} }});//map操作，找出用户的idJavaDStream<String> blackListuserIDBasedInBatchOnhistroy =blackListBasedOnHistory.map(new Function<Tuple2<String,Long>, String>() {public String call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubreturn v1._1.split("\t")[2];} });//有一个问题，数据可能重复，在一个partition里面重复，这个好办；//但多个partition不能保证一个用户重复，需要对黑名单的整个rdd进行去重操作。//rdd去重了，partition也就去重了，一石二鸟，一箭双雕// 找出了黑名单，下一步就写入黑名单数据库表中JavaDStream<String> blackListUniqueuserBasedInBatchOnhistroy = blackListuserIDBasedInBatchOnhistroy.transform(new Function<JavaRDD<String>, JavaRDD<String>>() {public JavaRDD<String> call(JavaRDD<String> rdd) throws Exception {// TODO Auto-generated method stubreturn rdd.distinct();} });// 下一步写入到数据表中blackListUniqueuserBasedInBatchOnhistroy.foreachRDD(new Function<JavaRDD<String>, Void>() {public Void call(JavaRDD<String> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<String>>() {public void call(Iterator<String> t) throws Exception {// TODO Auto-generated method stub//插入的用户信息可以只包含：useID//此时直接插入黑名单数据表即可。//写入数据库List<Object[]> blackList = new ArrayList<Object[]>();while(t.hasNext()) {blackList.add(new Object[]{t.next()});}JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doBatch("INSERT INTO blacklisttable values (?)", blackList);} });return null;} });/广告点击累计动态更新,每个updateStateByKey都会在Batch Duration的时间间隔的基础上进行广告点击次数的更新， 更新之后我们一般都会持久化到外部存储设备上，在这里我们存储到MySQL数据库中/JavaPairDStream<String, Long> updateStateByKeyDSteam = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} }).updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() {public Optional<Long> call(List<Long> v1, Optional<Long> v2)throws Exception {// v1:当前的Key在当前的Batch Duration中出现的次数的集合，例如{1，1，1，。。。，1}// v2:当前的Key在以前的Batch Duration中积累下来的结果；Long clickedTotalHistory = 0L; if(v2.isPresent()){clickedTotalHistory = v2.get();}for(Long one : v1) {clickedTotalHistory += one;}return Optional.of(clickedTotalHistory);} });updateStateByKeyDSteam.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：timestamp、adID、province、city//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<AdClicked> AdClickedList = new ArrayList<AdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");AdClicked adClicked = new AdClicked();adClicked.setTimestamp(splited[0]);adClicked.setAdID(splited[1]);adClicked.setProvince(splited[2]);adClicked.setCity(splited[3]);adClicked.setClickedCount(record._2);AdClickedList.add(adClicked);}final List<AdClicked> inserting = new ArrayList<AdClicked>();final List<AdClicked> updating = new ArrayList<AdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdClicked clicked : AdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedcount WHERE"+ " timestamp = ? AND adID = ? AND province = ? AND city = ?",new Object[]{clicked.getTimestamp(), clicked.getAdID(),clicked.getProvince(), clicked.getCity()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedcount VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.getTimestamp(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity()});}jdbcWrapper.doBatch("UPDATE adclickedcount SET clickedCount = ? WHERE"+ " timestamp =? AND adID = ? AND province = ? AND city = ?", updateParametersList);} });return null;} });/ 对广告点击进行TopN计算，计算出每天每个省份Top5排名的广告 因为我们直接对RDD进行操作，所以使用了transfomr算子；/updateStateByKeyDSteam.transform(new Function<JavaPairRDD<String,Long>, JavaRDD<Row>>() {public JavaRDD<Row> call(JavaPairRDD<String, Long> rdd) throws Exception {JavaRDD<Row> rowRDD = rdd.mapToPair(new PairFunction<Tuple2<String,Long>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, Long> t)throws Exception {// TODO Auto-generated method stubString[] splited=t._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];String clickedRecord = timestamp + "_" + adID + "_" + province;return new Tuple2<String, Long>(clickedRecord, t._2);} }).reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }).map(new Function<Tuple2<String,Long>, Row>() {public Row call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubString[] splited=v1._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];return RowFactory.create(timestamp, adID, province, v1._2);} });StructType structType = DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("timestamp", DataTypes.StringType, true),DataTypes.createStructField("adID", DataTypes.StringType, true),DataTypes.createStructField("province", DataTypes.StringType, true),DataTypes.createStructField("clickedCount", DataTypes.LongType, true)));HiveContext hiveContext = new HiveContext(rdd.context());DataFrame df = hiveContext.createDataFrame(rowRDD, structType);df.registerTempTable("topNTableSource");DataFrame result = hiveContext.sql("SELECT timestamp, adID, province, clickedCount, FROM"+ " (SELECT timestamp, adID, province,clickedCount, "+ "ROW_NUMBER() OVER(PARTITION BY province ORDER BY clickeCount DESC) rank "+ "FROM topNTableSource) subquery "+ "WHERE rank <= 5");return result.toJavaRDD();} }).foreachRDD(new Function<JavaRDD<Row>, Void>() {public Void call(JavaRDD<Row> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Row>>() {public void call(Iterator<Row> t) throws Exception {// TODO Auto-generated method stubList<AdProvinceTopN> adProvinceTopN = new ArrayList<AdProvinceTopN>();while(t.hasNext()) {Row row = t.next();AdProvinceTopN item = new AdProvinceTopN();item.setTimestamp(row.getString(0));item.setAdID(row.getString(1));item.setProvince(row.getString(2));item.setClickedCount(row.getLong(3));adProvinceTopN.add(item);}// final List<AdProvinceTopN> inserting = new ArrayList<AdProvinceTopN>();// final List<AdProvinceTopN> updating = new ArrayList<AdProvinceTopN>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();Set<String> set = new HashSet<String>();for(AdProvinceTopN item: adProvinceTopN){set.add(item.getTimestamp() + "_" + item.getProvince());}//表的字段timestamp、adID、province、clickedCountArrayList<Object[]> deleteParametersList = new ArrayList<Object[]>();for(String deleteRecord : set) {String[] splited = deleteRecord.split("_");deleteParametersList.add(new Object[]{splited[0],splited[1]});}jdbcWrapper.doBatch("DELETE FROM adprovincetopn WHERE timestamp = ? AND province = ?", deleteParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdProvinceTopN insertRecord : adProvinceTopN) {insertParametersList.add(new Object[] {insertRecord.getClickedCount(),insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince()});}jdbcWrapper.doBatch("INSERT INTO adprovincetopn VALUES (?, ?, ?, ?)", insertParametersList);} });return null;} });/ 计算过去半个小时内广告点击的趋势 广告点击的基本数据格式：timestamp、ip、userID、adID、province、city/filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String splited[] = t._2.split("\t");String adID = splited[3];String time = splited[0]; //Todo:后续需要重构代码实现时间戳和分钟的转换提取。此处需要提取出该广告的点击分钟单位return new Tuple2<String, Long>(time + "_" + adID, 1L);} }).reduceByKeyAndWindow(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }, new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 - v2;} }, Durations.minutes(30), Durations.milliseconds(5)).foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition)throws Exception {List<AdTrendStat> adTrend = new ArrayList<AdTrendStat>();// TODO Auto-generated method stubwhile(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("_");String time = splited[0];String adID = splited[1];Long clickedCount = record._2;/ 在插入数据到数据库的时候具体需要哪些字段？time、adID、clickedCount; 而我们通过J2EE技术进行趋势绘图的时候肯定是需要年、月、日、时、分这个维度的，所以我们在这里需要 年月日、小时、分钟这些时间维度；/AdTrendStat adTrendStat = new AdTrendStat();adTrendStat.setAdID(adID);adTrendStat.setClickedCount(clickedCount);adTrendStat.set_date(time); //Todo:获取年月日adTrendStat.set_hour(time); //Todo:获取小时adTrendStat.set_minute(time);//Todo:获取分钟adTrend.add(adTrendStat);}final List<AdTrendStat> inserting = new ArrayList<AdTrendStat>();final List<AdTrendStat> updating = new ArrayList<AdTrendStat>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdTrendStat trend : adTrend) {final AdTrendCountHistory adTrendhistory = new AdTrendCountHistory();jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedtrend WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?",new Object[]{trend.get_date(), trend.get_hour(), trend.get_minute(),trend.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);adTrendhistory.setClickedCountHistoryLong(count);updating.add(trend);} else { inserting.add(trend);} }});}//表的字段date、hour、minute、adID、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdTrendStat insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.get_date(),insertRecord.get_hour(),insertRecord.get_minute(),insertRecord.getAdID(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedtrend VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段date、hour、minute、adID、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdTrendStat updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.get_date(),updateRecord.get_hour(),updateRecord.get_minute(),updateRecord.getAdID()});}jdbcWrapper.doBatch("UPDATE adclickedtrend SET clickedCount = ? WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?", updateParametersList);} });return null;} });;/ Spark Streaming 执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于 接收应用程序本身或者Executor中的消息，/javassc.start();javassc.awaitTermination();javassc.close();}private static JavaStreamingContext createContext(String checkpointDirectory, SparkConf conf) {// If you do not see this printed, that means the StreamingContext has been loaded// from the new checkpointSystem.out.println("Creating new context");// Create the context with a 5 second batch sizeJavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(10));ssc.checkpoint(checkpointDirectory);return ssc;} }class JDBCWrapper {private static JDBCWrapper jdbcInstance = null;private static LinkedBlockingQueue<Connection> dbConnectionPool = new LinkedBlockingQueue<Connection>();static {try {Class.forName("com.mysql.jdbc.Driver");} catch (ClassNotFoundException e) {// TODO Auto-generated catch blocke.printStackTrace();} }public static JDBCWrapper getJDBCInstance() {if(jdbcInstance == null) {synchronized (JDBCWrapper.class) {if(jdbcInstance == null) {jdbcInstance = new JDBCWrapper();} }}return jdbcInstance; }private JDBCWrapper() {for(int i = 0; i < 10; i++){try {Connection conn = DriverManager.getConnection("jdbc:mysql://Master:3306/sparkstreaming","root", "root");dbConnectionPool.put(conn);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } }public synchronized Connection getConnection() {while(0 == dbConnectionPool.size()){try {Thread.sleep(20);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }return dbConnectionPool.poll();}public int[] doBatch(String sqlText, List<Object[]> paramsList){Connection conn = getConnection();PreparedStatement preparedStatement = null;int[] result = null;try {conn.setAutoCommit(false);preparedStatement = conn.prepareStatement(sqlText);for(Object[] parameters: paramsList) {for(int i = 0; i < parameters.length; i++){preparedStatement.setObject(i + 1, parameters[i]);} preparedStatement.addBatch();}result = preparedStatement.executeBatch();conn.commit();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }}return result; }public void doQuery(String sqlText, Object[] paramsList, ExecuteCallBack callback){Connection conn = getConnection();PreparedStatement preparedStatement = null;ResultSet result = null;try {preparedStatement = conn.prepareStatement(sqlText);for(int i = 0; i < paramsList.length; i++){preparedStatement.setObject(i + 1, paramsList[i]);} result = preparedStatement.executeQuery();try {callback.resultCallBack(result);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }} }}interface ExecuteCallBack {void resultCallBack(ResultSet result) throws Exception;}class UserAdClicked {private String timestamp;private String ip;private String userID;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public String getUserID() {return userID;}public void setUserID(String userID) {this.userID = userID;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdClicked {private String timestamp;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdProvinceTopN {private String timestamp;private String adID;private String province;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendStat {private String _date;private String _hour;private String _minute;private String adID;private Long clickedCount;public String get_date() {return _date;}public void set_date(String _date) {this._date = _date;}public String get_hour() {return _hour;}public void set_hour(String _hour) {this._hour = _hour;}public String get_minute() {return _minute;}public void set_minute(String _minute) {this._minute = _minute;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendCountHistory{private Long clickedCountHistoryLong;public Long getClickedCountHistoryLong() {return clickedCountHistoryLong;}public void setClickedCountHistoryLong(Long clickedCountHistoryLong) {this.clickedCountHistoryLong = clickedCountHistoryLong;} } 本篇文章为转载内容。原文链接：https://blog.csdn.net/tom_8899_li/article/details/71194434。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...应用。在本文中，通过实例演示了如何创建Vue实例、进行数据绑定、使用计算属性、条件渲染、列表渲染和事件处理等核心功能。 数据绑定 , 在Vue.js中，数据绑定是一种自动同步组件数据与DOM（文档对象模型）的技术。当数据发生变化时，Vue会自动更新相应的DOM元素，反之亦然。例如，在文章中，我们通过 message 实现将Vue实例中的message变量与页面上的文本内容进行绑定。 计算属性（Computed Properties） , Vue.js中的计算属性是对Vue实例内部数据的一种抽象封装，可以基于响应式依赖的数据生成新的衍生值。每当这些依赖的数据发生改变时，计算属性会自动重新计算并返回结果。文中提到的reverseMessage就是一个计算属性，它根据data对象中的message值实时计算并返回反转后的字符串。 条件渲染（v-if） , 在Vue.js中，条件渲染是通过v-if指令实现的，允许开发者根据表达式的真假来决定是否渲染或销毁元素。如示例代码所示，当show为真时，v-if=\ show\ 控制的段落元素会被渲染到页面上，否则不会显示。 列表渲染（v-for） , Vue.js中的列表渲染是通过v-for指令完成的，它可以遍历数组或对象，并依据其中的每一项动态生成多个DOM元素。文中展示的例子中，v-for=\ item in items\ 会针对items数组中的每个元素生成一个列表项（li），并将当前元素的值插入到列表项内。

... 在计算机视觉与光学字符识别（OCR）领域，Tesseract作为一款开源且功能强大的工具，被广泛应用。然而，在实际使用过程中，我们可能会遇到一些识别错误或异常情况，这时如何正确地理解和处理这些问题呢？本文将带你一起深入探讨，并通过实例代码来具体展示。 1. 理解Tesseract的局限性 首先，我们需要认识到即使是Tesseract这样的优秀OCR引擎，也无法做到100%准确。其性能受到图片质量、字体样式、背景复杂度等因素的影响。所以，当遇到识别出岔子的时候，咱首先别急着满世界找解决办法，而是要先稳住心态，理解和欣然接受这个实际情况。接下来，咱就可以对症下药，要么琢磨着优化一下输入的照片，要么灵活调整一下参数设定，这样就对啦！ python import pytesseract from PIL import Image 假设我们有一张较为复杂的图片需要识别 img = Image.open('complex_image.png') text = pytesseract.image_to_string(img) 如果输出的text有误，那可能是因为原始图片的质量问题 2. 图像预处理 为了提高识别准确性，对输入图像进行预处理是至关重要的一步。例如，我们可以进行灰度化、二值化、降噪、边界检测等操作。 python 对图片进行灰度化和二值化处理 img = img.convert('L').point(lambda x: 0 if x < 128 else 255, '1') 再次尝试识别 improved_text = pytesseract.image_to_string(img) 3. 调整识别参数 Tesseract提供了一系列丰富的可调参数以适应不同的场景。比如语言模型、是否启用特定字典、识别模式等。针对特定场景下的错误，可以通过调整这些参数来改善识别效果。 python 使用英语+数字的语言模型，同时启用多层识别 custom_config = r'--oem 3 --psm 6 -l eng' more_accurate_text = pytesseract.image_to_string(img, config=custom_config) 4. 结果后处理 即便进行了以上优化，识别结果仍可能出现瑕疵。这时候，我们可以灵活运用自然语言处理技术对结果进行深加工，比如纠错、分词、揪出关键词这些操作，这样一来，文本的实用性就能噌噌噌地往上提啦！ python import re from nltk.corpus import words 创建一个简单的英文单词库 english_words = set(words.words()) 对识别结果进行过滤，只保留英文单词 filtered_text = ' '.join([word for word in improved_text.split() if word.lower() in english_words]) 5. 针对异常情况的处理 当Tesseract抛出异常时，应遵循常规的异常处理原则。例如，捕获Image.open()可能导致的IOError，或者pytesseract.image_to_string()可能引发的RuntimeError等。 python try: img = Image.open('nonexistent_image.png') text = pytesseract.image_to_string(img) except IOError: print("无法打开图片文件！") except RuntimeError as e: print(f"运行时错误：{e}") 总结来说，处理Tesseract的错误和异常情况是一项涉及多个层面的工作，包括理解其内在局限性、优化输入图像、调整识别参数、结果后处理以及有效应对异常。在这个过程中，耐心调试、持续学习和实践反思都是非常关键的。让我们用人类特有的情感化思考和主观能动性去驾驭这一强大的工具，让Tesseract更好地服务于我们的需求吧！

...我们，我们将第一时间进行核实并删除相应内容。 在一个实际需求中，需要对一批文件(如：文本、图片)进行重命名，按照数字编号。正好借此熟悉了一下node的fs文件操作，写了一个批量修改文件名的脚本。 原文链接 需求 现有以下图片文件 修改前 需要批量修改文件名称，变为统一前缀名称并且自增索引，修改后效果 修改后 最简单的人力操作就是逐个文件重命名，但本着DRY(Don't repeat yourself)原则，还是写一个node脚本搞定。 研究 node中要进行文件操作需要了解一下fs模块 在fs模块中有同步和异步两种方式 读取文件 //异步 fs.readFile('test.txt', 'utf-8' (err, data) => { if (err) { throw err; } console.log(data); }); //同步 let data = fs.readFileSync('test.txt'); console.log(data); 异步读取文件参数：文件路径，编码方式，回调函数 写入文件 fs.writeFile('test2.txt', 'this is text', { 'flag': 'w' }, err => { if (err) { throw err; } console.log('saved'); }); 写入文件参数：目标文件，写入内容，写入形式，回调函数 flag写入方式： r：读取文件 w：写文件 a：追加 创建目录 fs.mkdir('dir', (err) => { if (err) { throw err; } console.log('make dir success'); }); dir为新建目录名称 读取目录 fs.readdir('dir',(err, files) => { if (err) { throw err; } console.log(files); }); dir为读取目录名称，files为目录下的文件或目录名称数组 获取文件信息 fs.stat('test.txt', (err, stats)=> { console.log(stats.isFile()); //true }) 获取文件信息后stats方法： 方法 说明 stats.isFile() 是否为文件 stats.isDirectory() 是否为目录 stats.isBlockDevice() 是否为块设备 stats.isCharacterDevice() 是否为字符设备 stats.isSymbolicLink() 是否为软链接 stats.isFIFO() 是否为UNIX FIFO命令管道 stats.isSocket() 是否为Socket 创建读取流 let stream = fs.createReadStream('test.txt'); 创建写入流 let stream = fs.createWriteStreamr('test_copy.txt'); 开发 开发思路： 读取源目录 判读存放目录是否存在，不存在时新建目录 复制文件 判断复制内容是否为文件 创建读取流 创建写入流 链接管道，写入文件内容 let fs = require('fs'), src = 'src', dist = 'dist', args = process.argv.slice(2), filename = 'image', index = 0; //show help if (args.length === 0 || args[0].match('--help')) { console.log('--help\n \t-src 文件源\n \t-dist 文件目标\n \t-n 文件名\n \t-i 文件名索引\n'); return false; } args.forEach((item, i) => { if (item.match('-src')) { src = args[i + 1]; } else if (item.match('-dist')) { dist = args[i + 1]; } else if (item.match('-n')) { filename = args[i + 1]; } else if (item.match('-i')) { index = args[i + 1]; } }); fs.readdir(src, (err, files) => { if (err) { console.log(err); } else { fs.exists(dist, exist => { if (exist) { copyFile(files, src, dist, filename, index); } else { fs.mkdir(dist, () => { copyFile(files, src, dist, filename, index); }) } }); } }); function copyFile(files, src, dist, filename, index) { files.forEach(n => { let readStream, writeStream, arr = n.split('.'), oldPath = src + '/' + n, newPath = dist + '/' + filename + index + '.' + arr[arr.length - 1]; fs.stat(oldPath, (err, stats) => { if (err) { console.log(err); } else if (stats.isFile()) { readStream = fs.createReadStream(oldPath); writeStream = fs.createWriteStream(newPath); readStream.pipe(writeStream); } }); index++; }) } 效果 总结 node提供了很多模块可以帮助我们完成不同需求的功能开发，使javascript不仅仅局限与浏览器中，尝试自己编写一些脚本有助于对这些模块的理解，同时也能提高办公效率。 本篇文章为转载内容。原文链接：https://blog.csdn.net/weixin_33205138/article/details/112036462。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...域，然后在这个区域内进行识别。这样就可以避免图像抖动和变形带来的影响。 python import cv2 image = cv2.imread('low_quality_image.png', 0) gray = cv2.medianBlur(image, 5) Otsu's thresholding after Gaussian filtering blur = cv2.GaussianBlur(gray,(5,5),0) _, thresh = cv2.threshold(blur, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU) contours, _ = cv2.findContours(thresh, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE) contours = sorted(contours, key=cv2.contourArea, reverse=True)[:5] for c in contours: x,y,w,h = cv2.boundingRect(c) roi_gray = gray[y:y+h, x:x+w] if cv2.countNonZero(roi_gray) < 100: continue cv2.rectangle(image,(x,y),(x+w,y+h),(255,0,0),2) cv2.imshow('Image', image) cv2.waitKey(0) cv2.destroyAllWindows() 3. 字符分割 对于模糊的问题，我们可以尝试字符分割的方法，即将图片中的每一个字符都单独提取出来，然后再分别进行识别。这样可以有效地避免整个图片识别错误的情况。 python import pytesseract from PIL import Image image = Image.open('low_quality_image.png') text = pytesseract.image_to_string(image) words = text.split() for word in words: word_image = image.crop((0, 0, len(word), 1)) print(pytesseract.image_to_string(word_image)) 四、结语 通过以上的分析和讨论，我们可以看出，虽然低质量图像给Tesseract的识别带来了一定的挑战，但是我们还是可以通过一系列的优化策略来提升其性能。真心希望这篇文章能给亲带来一些实实在在的帮助，如果有啥疑问、想法或者建议，尽管随时找我唠唠嗑，咱一起探讨探讨哈！

...转换和处理过程，通过实例代码揭示其背后的奥秘。 1. Hadoop生态系统简介 Hadoop的核心组件主要包括HDFS（Hadoop Distributed File System）和MapReduce。HDFS负责海量数据的分布式存储，而MapReduce则提供了并行处理大规模数据集的强大能力。在此基础上，我们可以通过编写特定的Map和Reduce函数，实现对原始数据的转换和处理。 2. 数据转换 Map阶段 让我们首先通过一个简单的示例理解Hadoop MapReduce中的数据转换过程： java import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class WordCountMapper extends Mapper { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); for (String eachWord : line.split("\\s+")) { word.set(eachWord); context.write(word, one); // 将单词作为key，计数值1作为value输出 } } } 这段代码是Hadoop实现词频统计任务的Mapper部分，它实现了数据从原始文本格式到键值对形式的转换。当Map阶段读取每行文本时，将其拆分为单个单词，并以单词为键、值为1的形式输出，实现了初步的数据转换。 3. 数据处理 Reduce阶段 接下来，我们看下Reduce阶段如何进一步处理这些键值对，完成最终的数据聚合： java import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WordCountReducer extends Reducer { public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); // 对所有相同键的值进行累加 } context.write(key, new IntWritable(sum)); // 输出每个单词及其出现次数 } } 在上述Reducer类中，对于每一个输入的单词（键），我们将所有关联的计数值（值）相加，得到该单词在整个文本中的出现次数，从而完成了数据的聚合处理。 4. 思考与讨论 Hadoop的魅力在于，通过分解复杂的计算任务为一系列简单的Map和Reduce操作，我们可以轻松地应对海量数据的转换和处理。这种并行计算模型就像是给电脑装上了超级引擎，让数据处理速度嗖嗖地往上窜。而且更棒的是，它把数据分散存放在一整个集群的各个节点上，就像把鸡蛋放在不同的篮子里一样。这样一来，不仅能够轻松应对大规模运算，就算某个节点出个小差错，其他的节点也能稳稳接住，保证整个系统的稳定性和可扩展性杠杠的！ 然而，尽管Hadoop在数据处理方面表现出色，但并非所有场景都适用。比如，在那种需要迅速反馈或者频繁做大量计算的情况下，像Spark这类流处理框架或许会是个更棒的选择。这就意味着在咱们实际操作的项目里，面对不同的需求和技术特点时，咱们得像个精明的小侦探，灵活机智地挑出最对味、最适合的数据处理武器和战术方案。 总的来说，借助Hadoop，我们能够构建出高效的数据转换和处理流程，从容应对大数据挑战。不过呢，咱们也得时刻想着把它的原理摸得更透彻些，还有怎么跟其他的技术工具灵活搭配使用。这样一来，咱就能在那些乱七八糟、变来变去的业务环境里头，发挥出更大的作用，创造更大的价值啦！

...我们使用read函数从指定目录下读取CSV文件，并将其转化为DataFrame。然后，我们可以通过各种函数对DataFrame进行操作，如show、filter、groupBy等。 2. 使用Spark SQL Spark SQL是一种高级抽象，用于查询关系表。就像Dataframe API那样，Spark SQL也给我们带来了一种超级实用又高效的处理小文件的方法，一点儿也不复杂，特别接地气儿。Spark SQL还自带了一堆超级实用的内置函数，比如COUNT、SUM、AVG这些小帮手，用它们来处理小文件，那速度可真是嗖嗖的，轻松又高效。 下面是一个简单的例子，展示如何使用Spark SQL来读取小文件： scss val df = spark.sql("SELECT FROM /path/to/files/") 在这个例子中，我们使用sql函数来执行SQL语句，从而从指定目录下读取CSV文件并转化为DataFrame。 3. 使用Partitioner Partitioner是Spark的一种内置机制，用于将数据分割成多个块。当我们处理大量小文件时，可以使用Partitioner来提高处理效率。其实呢，我们可以这样来操作：比如说，按照文件的名字呀，或者文件里边的内容这些规则，把那些小文件分门别类地整理一下。就像是给不同的玩具放在不同的抽屉里一样，每个类别都单独放到一个文件夹里面去存储，这样一来就清清楚楚、井井有条啦！这样一来，每次我们要读取文件的时候，就只需要瞄一眼一个文件夹里的内容，压根不需要把整个目录下的所有文件都翻个底朝天。 下面是一个简单的例子，展示如何使用Partitioner来处理小文件： python val partitioner = new HashPartitioner(5) val rdd = sc.textFile("/path/to/files/") .map(line => (line.split(",").head, line)) .partitionBy(partitioner) val output = rdd.saveAsTextFile("/path/to/output/") 在这个例子中，我们首先使用textFile函数从指定目录下读取文本文件，并将其转化为RDD。接着，我们运用一个叫做map的神奇小工具，就像魔法师挥动魔杖那样，把每一行文本巧妙地一分为二，一部分是文件名，另一部分则是内容。然后，我们采用了一个叫做partitionBy的神奇函数，就像把RDD里的数据放进不同的小篮子里那样，按照文件名给它们分门别类。这样一来，每个“篮子”里都恰好装了5个小文件，整整齐齐，清清楚楚。最后，我们使用saveAsTextFile函数将RDD保存为文本文件。因为我们已经按照文件名把文件分门别类地放进不同的“小桶”里了，所以现在每次找文件读取的时候，就不用像无头苍蝇一样满目录地乱窜，只需要轻轻松松打开一个文件夹，就能找到我们需要的文件啦！ 四、结论 通过以上三种方法，我们可以有效地优化Spark在读取大量小文件时的性能。Dataframe API和Spark SQL提供了简单且高效的API，可以快速处理结构化数据。Partitioner这个小家伙，就像个超级有条理的文件整理员，它能够按照特定的规则，麻利地把那些小文件分门别类放好。这样一来，当你需要读取文件的时候，就仿佛拥有了超能力一般，嗖嗖地提升读取速度，让效率飞起来！当然啦，这只是入门级别的小窍门，真正要让方案火力全开，还得瞅准实际情况灵活变通，不断打磨和优化才行。

...我们，我们将第一时间进行核实并删除相应内容。 SFTP是Secure File Transfer Protocol的缩写，安全文件传送协议。可以为传输文件提供一种安全的加密方法。SFTP 为 SSH的一部份，是一种传输文件到服务器的安全方式。SFTP是使用加密传输认证信息和传输的数据，所以，使用SFTP是非常安全的。但是，由于这种传输方式使用了加密/解密技术，所以传输效率比普通的FTP要低得多，如果您对网络安全性要求更高时，可以使用SFTP代替FTP(来自百度的解释) JSch是Java Secure Channel的缩写。 JSch是一个SSH2的纯Java实现。它允许你连接到一个SSH服务器，并且可以使用端口转发，X11转发，文件传输等，当然你也可以集成它的功能到你自己的应用程序。 ChannelSftp类是JSch实现SFTP核心类，它包含了所有SFTP的方法，如： put()： 文件上传get()： 文件下载cd()： 进入指定目录ls()： 得到指定目录下的文件列表rename()： 重命名指定文件或目录rm()： 删除指定文件mkdir()： 创建目录rmdir()： 删除目录 1、先引入jar包 <dependency><groupId>org.mybatis</groupId><artifactId>mybatis-spring</artifactId><version>1.2.2</version></dependency> 账号密码类 public interface SFTPDTO {/FTP登录用户名/public static final String username=xxxx;/ FTP登录密码/public static final String password=xxxx;/ 私钥/public static final String privateKey = xxxx;/ FTP服务器地址IP地址/public static final String host=xxxx;/ FTP端口/public static final int port=xxxx;} 重要类，里面包含开启连接和关闭连接。 public class SFTPUtils {private ChannelSftp sftp;private Session session;public void login(){try {JSch jsch = new JSch();if (SFTPDTO.privateKey != null) {jsch.addIdentity(SFTPDTO.privateKey);// 设置私钥}session = jsch.getSession(SFTPDTO.username, SFTPDTO.host, SFTPDTO.port);if (SFTPDTO.password != null) {session.setPassword(SFTPDTO.password);}Properties config = new Properties();config.put("StrictHostKeyChecking", "no");session.setConfig(config);session.connect();Channel channel = session.openChannel("sftp");channel.connect();sftp = (ChannelSftp) channel;} catch (Exception e) {log.error("Cannot connect to specified sftp server : {}:{} \n Exception message is: {}", new Object[]{SFTPDTO.host, SFTPDTO.port, e.getMessage()});} }/ 关闭连接 server/public void logout(){if (sftp != null) {if (sftp.isConnected()) {sftp.disconnect();log.info("sftp is closed already");} }if (session != null) {if (session.isConnected()) {session.disconnect();log.info("sshSession is closed already");} }}/ 将输入流的数据上传到sftp作为文件 @param directory 上传到该目录 @param sftpFileName sftp端文件名 @throws SftpException @throws Exception/public void upload(String directory, String sftpFileName, InputStream input) throws SftpException{try {sftp.cd(directory);} catch (SftpException e) {log.warn("directory is not exist");sftp.mkdir(directory);sftp.cd(directory);}sftp.put(input, sftpFileName);log.info("file:{} is upload successful" , sftpFileName);} } 测试一下 public static void main(){SFTPUtils sftp = new SFTPUtils();sftp.login();String audioUrl = courseSection.getAudioUrl();String temp[] = audioUrl.split("\\\\");String fileName = temp[temp.length - 1];InputStream inputStream = FileUtils.urlInputStream(audioUrl);sftp.upload("/www/website/haha/audio", fileName, inputStream);//上传//拼接最终的urlString newUrl = "https://static.taobao.com/website/ancai/audio/".concat(fileName);sftp.logout();} 把url转成流 public class FileUtils {public static InputStream urlInputStream(String fileUrl){if(StringUtils.isBlank(fileUrl)){return null;}try {URL url = new URL(fileUrl);HttpURLConnection conn = (HttpURLConnection)url.openConnection();//设置超时间为3秒conn.setConnectTimeout(31000);//防止屏蔽程序抓取而返回403错误conn.setRequestProperty("User-Agent", "Mozilla/4.0 (compatible; MSIE 5.0; Windows NT; DigExt)");//得到输入流return conn.getInputStream();} catch (Exception e) {//打印errorlog.error("fileutils.urlinputstream-获取url流失败:",e.getMessage());}return null;} } 实际中，我们使用这个工具类就够用了 public class SFTPUtils {private ChannelSftp sftp;private Session session;public void login(){try {JSch jsch = new JSch();if (SFTPDTO.privateKey != null) {jsch.addIdentity(SFTPDTO.privateKey);// 设置私钥}session = jsch.getSession(SFTPDTO.username, SFTPDTO.host, SFTPDTO.port);if (SFTPDTO.password != null) {session.setPassword(SFTPDTO.password);}Properties config = new Properties();config.put("StrictHostKeyChecking", "no");session.setConfig(config);session.connect();Channel channel = session.openChannel("sftp");channel.connect();sftp = (ChannelSftp) channel;} catch (Exception e) {log.error("Cannot connect to specified sftp server : {}:{} \n Exception message is: {}", new Object[]{SFTPDTO.host, SFTPDTO.port, e.getMessage()});} }/ 关闭连接 server/public void logout(){if (sftp != null) {if (sftp.isConnected()) {sftp.disconnect();log.info("sftp is closed already");} }if (session != null) {if (session.isConnected()) {session.disconnect();log.info("sshSession is closed already");} }}/ 将输入流的数据上传到sftp作为文件 @param directory 上传到该目录 @param sftpFileName sftp端文件名 @throws SftpException @throws Exception/public void upload(String directory, String sftpFileName, InputStream input) throws SftpException{try {sftp.cd(directory);} catch (SftpException e) {log.warn("directory is not exist");sftp.mkdir(directory);sftp.cd(directory);}sftp.put(input, sftpFileName);log.info("file:{} is upload successful" , sftpFileName);}/ 上传单个文件 @param directory 上传到sftp目录 @param uploadFile 要上传的文件,包括路径 @throws FileNotFoundException @throws SftpException @throws Exception/public void upload(String directory, String uploadFile) throws FileNotFoundException, SftpException{File file = new File(uploadFile);upload(directory, file.getName(), new FileInputStream(file));}/ 将byte[]上传到sftp，作为文件。注意:从String生成byte[]是，要指定字符集。 @param directory 上传到sftp目录 @param sftpFileName 文件在sftp端的命名 @param byteArr 要上传的字节数组 @throws SftpException @throws Exception/public void upload(String directory, String sftpFileName, byte[] byteArr) throws SftpException{upload(directory, sftpFileName, new ByteArrayInputStream(byteArr));}/ 将字符串按照指定的字符编码上传到sftp @param directory 上传到sftp目录 @param sftpFileName 文件在sftp端的命名 @param dataStr 待上传的数据 @param charsetName sftp上的文件，按该字符编码保存 @throws UnsupportedEncodingException @throws SftpException @throws Exception/public void upload(String directory, String sftpFileName, String dataStr, String charsetName) throws UnsupportedEncodingException, SftpException{upload(directory, sftpFileName, new ByteArrayInputStream(dataStr.getBytes(charsetName)));}/ 下载文件 @param directory 下载目录 @param downloadFile 下载的文件 @param saveFile 存在本地的路径 @throws SftpException @throws Exception/public void download(String directory, String downloadFile, String saveFile) throws SftpException, FileNotFoundException{if (directory != null && !"".equals(directory)) {sftp.cd(directory);}File file = new File(saveFile);sftp.get(downloadFile, new FileOutputStream(file));log.info("file:{} is download successful" , downloadFile);}/ 下载文件 @param directory 下载目录 @param downloadFile 下载的文件名 @return 字节数组 @throws SftpException @throws Exception/public byte[] download(String directory, String downloadFile) throws SftpException, IOException {if (directory != null && !"".equals(directory)) {sftp.cd(directory);}InputStream is = sftp.get(downloadFile);byte[] fileData = IOUtils.toByteArray(is);log.info("file:{} is download successful" , downloadFile);return fileData;}/ 删除文件 @param directory 要删除文件所在目录 @param deleteFile 要删除的文件 @throws SftpException @throws Exception/public void delete(String directory, String deleteFile) throws SftpException{sftp.cd(directory);sftp.rm(deleteFile);}/ 列出目录下的文件 @param directory 要列出的目录 @return @throws SftpException/public Vector<?> listFiles(String directory) throws SftpException {return sftp.ls(directory);}/public static void main(String[] args) throws SftpException, Exception {SFTPUtils sftp = new SFTPUtils("xxxx", "xxx", "upload.haha.com", 8888);sftp.login();InputStream inputStream = getInputStream("http://qiniu.xinxuanhaoke.com/keqianduwu_1.jpg");sftp.upload("/www/website/ancai/audio", "123.jpg", inputStream);sftp.logout();}/} 方式二、使用HuTool的工具类 先引入jar <dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.4.0</version></dependency><dependency><groupId>com.jcraft</groupId><artifactId>jsch</artifactId><version>0.1.53</version></dependency> public static void main(String[] args) {Sftp sftp = JschUtil.createSftp("ip或者域名", 端口, "账号", "密码");ChannelSftp client = sftp.getClient();String cd = "/www/website/ancai/audio";//要上传的路径try {sftp.cd(cd); //进入指定目录} catch (Exception e) {log.warn("directory is not exist");sftp.mkdir(cd); //创建目录sftp.cd(cd); //进入目录}InputStream inputStream = urlInputStream("http://audio.xinxuanhaoke.com/50bda079e9ef3673bbaeda20321bf932.mp3");//将文件转成流client.put(String.valueOf(inputStream), "1.mp3");//开始上传。} 本文引自：https://www.cnblogs.com/ceshi2016/p/7519762.html 本篇文章为转载内容。原文链接：https://blog.csdn.net/weixin_37862824/article/details/113530683。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...我们，我们将第一时间进行核实并删除相应内容。 首先感谢以下大佬提供帮助 斗鱼视频下载-> https://www.jianshu.com/p/feccccb097be 批量合并处理B站视频->https://www.wandouip.com/t5i227224/ You-Get->https://you-get.org/ 思路 首先打开了PC端斗鱼视频一个链接：https://v.douyu.com/show/0Q8mMY0xXDL749Ad 发现一个参数在链接中 0Q8mMY0xXDL749Ad 通过抓包发现了一个很重要的文件：playlist.m3u8 里面包含了数个 .ts 的网络地址； .ts 文件是可以播放的视频片段； 发现 可以通过合并 .ts 片段可以得到完整视频； 出现一个问题：playlist.m3u8 怎么获取？ 发现：https://v.douyu.com/api/stream/getStreamUrl 可以获取 playlist.m3u8 文件地址； 需要POST传入一些参数才行，发现： sign 参数是一种签名，一般通过JS生成，找了半天没有方法生成 sign 参数； 通过查阅大佬文献发现：手机端的斗鱼视频有接口可以直接获取 playlist.m3u8 文件地址，成功越过 sign 签名防线； 手机端斗鱼视频链接：https://vmobile.douyu.com/show/0Q8mMY0xXDL749Ad 通过抓包发现：https://vmobile.douyu.com/video/getInfo?vid=0Q8mMY0xXDL749Ad； 这就解决了playlist.m3u8 文件获取问题：json[‘data’][‘video_url’] 第一个难题解决！！； 综上所述，整理一下具体采集流程： 获取vid = 0Q8mMY0xXDL749Ad (就是链接中的参数)； 通过 https://vmobile.douyu.com/video/getInfo?vid=0Q8mMY0xXDL749Ad 获取 playlist.m3u8 文件地址； 解析 playlist.m3u8 文件提取所有 .ts文件； 下载所有 .ts 文件； 合并 .ts 成视频文件输出； Python实现 不要开启线程池，因为会有一些问题 app.py config 中可以配置 import requestsimport reimport jsonimport timeimport pymongoimport psutilfrom hashlib import md5from moviepy.editor import from multiprocessing import Pool基本配置config = {'UID':'gKpdxKRWXwaW',用户ID'CID':104,栏目ID'TYPE':1, 1=>按用户id采集列表，2=>按栏目ID采集列表'TIME_START':1,起始时间'TIME_ENT':500,结束时间'PAGE_START':1,起始页'PAGE_END':10,结束页'TIME_GE':0,每个下载间隔时间'POOL':False,是否开启线程池'CHECKID':True, True 过滤已经下载过的视频 False 不过滤'FILE_PATH':'F:/ceshi/',下载目录，【会自动创建文件夹】'TS_PATH':'F:/ceshi/download/',缓存文件目录，【会自动创建文件夹】'DB_URL':'localhost',数据库地址'DB_NAME':'douyu',数据库名称''DB_TABLE':'douyu'数据库表}MongoDB初始化client = pymongo.MongoClient(config['DB_URL'])mango_db = client[config['DB_NAME']]MongoDB存储def save_to_mango(result):if mango_db[config['DB_TABLE']].insert_one({'vid':result}):print('成功存储到MangoDB')return Truereturn FalseMongoDB验证重复def check_to_mongo(vid):count = mango_db[config['DB_TABLE']].find({'vid':vid}).count()if count==0:return Falsereturn True删除文件def del_file(page):if os.path.exists(page): 删除文件，可使用以下两种方法。os.remove(page) os.unlink(my_file)else:print('no such file:%s' % page)循环列表删除文件def loop_del_file(arr):for item in arr:del_file(item)请求器def get_content_requests(url):headers = {}headers['user-agent']='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.131 Safari/537.36'headers['cookie'] = 'dy_did=07f83a57d1d2e22942e0883200001501; acf_did=07f83a57d1d2e22942e0883200001501; Hm_lvt_e99aee90ec1b2106afe7ec3b199020a7=1556514266,1557050422,1557208315; acf_auth=; acf_auth_wl=; acf_uid=; acf_nickname=; acf_username=; acf_own_room=; acf_groupid=; acf_notification=; acf_phonestatus=; _dys_lastPageCode=page_video,page_video; Hm_lpvt_e99aee90ec1b2106afe7ec3b199020a7=1557209469; _dys_refer_action_code=click_author_video_cate2'try:req_content = requests.get(url,headers = headers)if req_content.status_code == 200:return req_contentprint('请求失败：',url)return Noneexcept:print('请求失败：', url)return None把时间换算成秒def str_to_int(time):try:time_array = time.split(':')time_int = (int(time_array[0])60)+int(time_array[1])return time_intexcept:print('~~~~~计算视频时间失败~~~~~')return None提取需要采集的数据def get_list(html,type = 1):data = []try:list_json = json.loads(str(html))for om in list_json['data']['list']:gtime = str_to_int(om['video_str_duration'])if gtime > config['TIME_START'] and gtime < config['TIME_ENT']:if type == 2:data.append({'title': om['title'], 'vid': om['url'].split('show/')[1]})else:data.append({'title': om['title'], 'vid': om['hash_id']})return dataexcept:print('~~~~~数据提取失败~~~~~')return None解析playlist.m3u8def get_ts_list(m3u8):data = []try:html_m3u8_json = json.loads(m3u8)m3u8_text = get_content_requests(html_m3u8_json['data']['video_url'])m3u8_vurl =html_m3u8_json['data']['video_url'].split('playlist.m3u8?')[0]if m3u8_text:get_text = re.findall(',\n(.?).ts(.?)\n',m3u8_text.text,re.S)for item in get_text:data.append(m3u8_vurl+item[0]+'.ts'+item[1])return datareturn Noneexcept:print('~~~~~解析playlist.m3u8失败~~~~~')return None 杀死moviepy产生的特定进程def killProcess(): 处理python程序在运行中出现的异常和错误try: pids方法查看系统全部进程pids = psutil.pids()for pid in pids: Process方法查看单个进程p = psutil.Process(pid) print('pid-%s,pname-%s' % (pid, p.name())) 进程名if p.name() == 'ffmpeg-win64-v4.1.exe': 关闭任务 /f是强制执行，/im对应程序名cmd = 'taskkill /f /im ffmpeg-win64-v4.1.exe 2>nul 1>null' python调用Shell脚本执行cmd命令os.system(cmd)except:pass下载.ts文件def download_ts(m3u8_list,name):try:if not os.path.exists(config['FILE_PATH']):os.makedirs(config['FILE_PATH'])if not os.path.exists(config['TS_PATH']):os.makedirs(config['TS_PATH'])if os.path.exists(config['FILE_PATH']+name+'.mp4'):name = name+'_'+str(int(time.time()))print('开始下载：',name)L = []R = []for p in m3u8_list:ts_find = get_content_requests(p)file_ts = '{0}{1}.ts'.format(config['TS_PATH'],md5(ts_find.content).hexdigest())with open(file_ts,'wb') as f:f.write(ts_find.content)R.append(file_ts)hebing = VideoFileClip(file_ts)L.append(hebing)killProcess()print('下载完成：',file_ts)mp4file = '{0}{1}.mp4'.format(config['FILE_PATH'],name)final_clip = concatenate_videoclips(L)final_clip.to_videofile(mp4file, fps=24, remove_temp=True)killProcess()loop_del_file(R)print('\n下载完成：',name)print('')return Trueexcept:print('~~~~~合成.ts文件失败~~~~~')return None下载视频列表def list_get_kong(list_json):for item in list_json:y = Trueif config['CHECKID']:if check_to_mongo(item['vid']):print('~~~~~检测到重复项~~~~~')y = Falseif y:get_show_html = get_content_requests('https://vmobile.douyu.com/video/getInfo?vid=' + item['vid'])if get_show_html:m3u8_list = get_ts_list(get_show_html.text)if m3u8_list:download = download_ts(m3u8_list, item['title'])if download: save_to_mango(item['vid'])time.sleep(config['TIME_GE'])控制器def main(page):if config['TYPE']==1:print('~~~~~按用户ID采集~~~~~')listurl = 'https://v.douyu.com/video/author/getAuthorVideoListByNew?up_id={0}&cate2_id=0&limit=30&page={1}'.format(config['UID'],page)get_list_html = get_content_requests(listurl)if get_list_html:list_json = get_list(get_list_html.text,1)if list_json:list_get_kong(list_json)else:print('~~~~~按列表ID采集~~~~~')listurl = 'https://v.douyu.com/video/video/listData?page={1}&cate2Id={0}&action=new'.format(config['CID'],page)get_list_html = get_content_requests(listurl)if get_list_html:list_json = get_list(get_list_html.text,2)if list_json:list_get_kong(list_json)初始化if __name__=='__main__':if config['POOL']:groups = [x for x in range(config['PAGE_START'],config['PAGE_END']+1)]pool = Pool()pool.map(main, groups)else:for item in range(config['PAGE_START'],config['PAGE_END']+1):main(item)print('~~~~~已经完成【所有操作】~~~~~') 总结：众所周知，BiliBili是一个学习的网站！ 本篇文章为转载内容。原文链接：https://blog.csdn.net/qq_35875470/article/details/89857445。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...所有查询参数，并将其解析为便于处理的JSON对象。同时，支持将新的参数集合或已更新的参数重新编码为URL查询字符串，进而实时更改浏览器地址栏中的URL状态，而无需刷新整个页面。此外，url.js还具备删除指定URL参数的能力，使得维护页面历史记录及实现平滑的状态管理变得更为便捷。总之，这款插件是构建现代Web应用程序时不可或缺的工具之一，它可以有效提升开发效率并增强用户体验，特别是在那些依赖于URL进行路由导航的应用中。 点我下载 文件大小：294.72 KB 您将下载一个JQuery插件资源包，该资源包内部文件的目录结构如下： 本网站提供JQuery插件下载功能，旨在帮助广大用户在工作学习中提升效率、节约时间。 本网站的下载内容来自于互联网。如您发现任何侵犯您权益的内容，请立即告知我们，我们将迅速响应并删除相关内容。 免责声明：站内所有资源仅供个人学习研究及参考之用，严禁将这些资源应用于商业场景。 若擅自商用导致的一切后果，由使用者承担责任。

...lass绑定，并通过实例代码展示了与旧版Vue的差异和优势。此外，文章还探讨了v-bind:class结合模板refs、计算属性以及组合式API（Composition API）等Vue高级特性的应用场景，帮助开发者提升组件化开发效率。 另外，InfoQ的一篇报道《Vue.js在大型项目中的CSS类管理策略》也值得一看，文中不仅回顾了v-bind:class的基本用法，还分享了一些实际项目中如何结合模块化、CSS预处理器等工具进行复杂场景下的class动态管理，这对于面临大规模应用架构挑战的前端开发者具有很高的参考价值。 最后，Vue社区的一些教程如"Vue Conditional Classes: The Complete Guide"，提供了大量实战案例，引导读者逐步掌握条件class的各种绑定技巧，包括但不限于基于状态切换、事件驱动、以及与其他Vue指令如v-if、v-for等协同工作的方法，为读者深化Vue技能树提供有力支撑。

...据交换的标准格式，其解析与操作的重要性日益凸显。 例如，2023年5月，Node.js社区发布了一篇关于优化JSON性能的文章，其中详述了如何利用最新版本V18中的JSON.parse()方法的新特性提高大数据量JSON解析速度。通过引入新的Streaming API和改进的内存管理机制，开发者可以更高效地处理大规模JSON数据流，并动态获取嵌套数组或对象的名字及其对应值。 另外，对于那些需要深度访问JSON结构的应用场景，如配置文件解析、复杂状态管理等，JavaScript提供了诸如Lodash这样的工具库，提供了诸如_.get()、_.set()等便捷方法，使得根据路径字符串动态获取或设置JSON任意层级的数据成为可能，大大提升了开发效率及代码可读性。 同时，针对安全性考量，在实际项目中处理JSON时应确保进行有效的数据验证和错误处理，防止因恶意构造或意外损坏的JSON数据导致的安全漏洞。例如，使用AJV等JSON Schema验证库，可以在数据解析前对其进行严格校验，从而降低潜在风险。 综上所述，对JSON数组名值获取的基础理解是前端乃至全栈开发者的必备技能之一，而随着技术发展和安全需求的提升，掌握更多先进的JSON处理策略与工具将为开发者应对各种复杂应用场景提供有力支持。

...触发时，会执行Vue实例中定义的方法。例如，在文中提到的v-on:keydown.enter表示当input元素触发keydown事件且键值为Enter时，会调用指定的方法（如handleEnter）进行处理。 methods属性 , methods是Vue实例对象的一个属性，它是一个包含各种方法的对象。在Vue中，methods用于组织和封装视图逻辑或与用户交互相关的函数。在本文场景下，methods对象中定义了一个名为handleEnter的方法，该方法会在用户按下input框内的回车键时被触发，并在此处编写处理回车键事件的具体代码逻辑。

...位、十位和百位等数位进行快速提取。 数据分析 , 数据分析是利用统计学和计算机科学技术对大量数据进行清洗、转换、探索性研究以及模型构建的过程，旨在揭示隐藏在数据中的信息，提供决策支持。在本文的语境下，Python作为实现数据分析的重要工具之一，能够借助内置函数或第三方库（如Pandas）高效处理和解析数据，包括但不限于数值分段、计算和转换等操作。 网络爬虫 , 网络爬虫是一种自动获取网页内容的程序，它通过模拟人类浏览器的行为，按照一定的规则自动遍历互联网上的信息资源，从而抓取并存储所需的数据。在网络爬虫技术的应用场景中，Python因其灵活、高效的特性，常被用来编写爬虫脚本以批量收集网络数据。文中提及Python的数学运算能力也能帮助开发者在抓取数据后对数字信息进行实时处理和格式转换，满足特定业务需求。

...简洁明了、便于阅读和解析，越来越受到开发者的欢迎。但是，如果在应用开发中经常对JSON进行操作，人工编写相关代码则会很复杂麻烦且易于出错。 为了提升开发效能，我们通常会应用JSON功能组件包中的功能组件，其中比较常用的功能组件是JSON Util。JSON Util是一款小巧轻便的Java JSON解析库，具有解析效率高、简单易学等优点。 下面我们简单介绍一下JSON Util的应用方法： //导入JSON Util库 import org.json.; //将JSON字符串解析成JSONObject String jsonStr = "{\"name\":\"Jack\",\"age\":\"25\",\"address\":{\"province\":\"Guangdong\",\"city\":\"Shenzhen\",\"district\":\"Nanshan\"} }"; JSONObject jsonObj = new JSONObject(jsonStr); //获取JSONObject中的某个字段 String name = jsonObj.getString("name"); //将JSONObject转换为Java Bean Person person = jsonObj.toJavaObject(Person.class); 通过调用JSON Util提供的API，我们可以轻松地从字符串中解析出JSON对象，并且获取JSON中的字段值或将其转换为Java Bean，这样就可以更方便地完成与JSON相关的开发任务了。 总之，JSON Util是一款非常实用的JSON解析库，如果你在Java开发中需要对JSON数据进行操作的话，就不妨尝试一下这个功能组件。

...cript引擎）直接解析和生成，广泛应用于Web服务接口、配置文件、数据存储等领域中，实现不同系统间的数据交换。 Python字典 , 在Python编程语言中，字典是一种可变容器模型，且可存储任意类型对象（如字符串、数字、元组等）的无序集合。每个项由一个唯一的键和与之关联的值组成，键和值之间通过冒号分隔，各对键值之间用逗号分隔，并放在花括号内。在处理JSON数据时，JSON对象通常会被转换为Python字典，以便在Python程序内部进行操作和处理。 JSON注入 , 类似于SQL注入，JSON注入是一种安全漏洞，攻击者通过向应用提交恶意构造的JSON数据，利用目标应用程序未能有效验证或清理用户输入的问题，进而影响服务器端JSON解析器的行为，以达到篡改数据、执行非预期操作或获取敏感信息的目的。在Python中使用json.loads()方法解析JSON字符串时，应结合严格的数据验证和清理措施来防止JSON注入攻击。

...种数据结构，通过哈希函数将输入（键）转换为数组索引以实现快速访问。在Java的HashMap和HashSet中，哈希表被用于存储元素，通过计算键的哈希值来确定元素在数组中的位置，从而支持O(1)时间复杂度的查找、插入和删除操作。当多个键映射到同一个索引时，会发生哈希冲突，可以通过链地址法、开放寻址法等方式解决。 哈希函数 , 哈希函数是一种特殊的函数，它可以将任意长度的输入（通常是字符串或对象）转化为固定长度的输出，即哈希值。在Java的HashMap和HashSet中，哈希函数用于计算键的哈希码，该哈希码随后用于确定元素在哈希表中的存储位置。一个好的哈希函数应该尽可能均匀地分布不同输入的哈希值，以减少哈希冲突并提高性能。 O(1)时间复杂度 , 在计算机科学中，O(1)时间复杂度表示算法的运行时间与问题规模无关，即无论输入数据大小如何，其执行时间恒定。在讨论HashMap和HashSet时，查找、插入和删除操作的时间复杂度都是O(1)，意味着只要哈希函数设计合理且哈希表负载均衡，对集合进行这些基本操作所需的时间是常数级别的，不随集合内元素数量的增长而增长。但这需要假设哈希函数将键均匀分布到桶中，并且桶的数量足够大以保持较低的冲突率。

...通过注解的方式，可以指定JSON字段与Java类属性之间的映射关系，使得转换过程更加智能且灵活。此外，对于可能存在的空值或异常情况，Jackson也提供了多种配置选项供开发者进行容错处理。 另一方面，Gson、Fastjson等其他开源库也是处理JSON与Java对象互转的有效工具，各有优劣，开发者可以根据项目需求和性能指标选择合适的库。同时，最新的Java版本（如Java 11及以上）已经原生支持JSON操作，例如使用JsonParser解析JSON，或者通过内置的JSON-B实现进行序列化和反序列化。 值得注意的是，在处理大量数据或高并发场景时，对JSON转换性能的优化至关重要。这包括但不限于选择高效的JSON库、合理设计数据模型以减少转换开销、利用缓冲技术提高IO效率等手段。因此，深入理解并掌握这些技术点，不仅能够提升程序性能，也能更好地应对实际开发中的各种挑战。

...的XSS攻击，强调在解析JSON时进行严格的数据类型检查与过滤。同时，ECMAScript（ES）规范也在不断演进，例如ES2019引入了JSON.stringify()方法的可选参数replacer，它能提供更为精细的控制，使得开发者能够选择性地序列化JSON对象中的特定属性。 另外，在前端框架React、Vue.js中，JSON数据的处理也得到了深度集成，通过状态管理库如Redux、Vuex，开发者可以便捷地操作JSON数据并在组件间传递。这些框架通常会提供工具函数或API，简化对JSON数据的遍历、更新和响应式绑定。 综上所述，掌握JavaScript中遍历JSON数据的基本技巧仅仅是开始，了解JSON在现代Web开发中的前沿应用、安全风险以及相关技术的发展趋势，对于提升项目质量和开发效率具有重要意义。

...hon中如何正确判断字符串相等之后，我们可以进一步探讨Python编程实践中关于字符串处理的其他关键议题。最近，Python 3.9版本引入了新的字符串方法removeprefix()和removesuffix()，这两个方法可以更高效地移除字符串开头或结尾的指定子串，提升了代码的简洁性和可读性。例如，在处理URL路径或者文件名时，常需要去除特定前缀或后缀，新方法提供了更为便捷的操作手段。 另外，对于大型应用中的字符串性能优化问题，推荐阅读关于Python内部字符串池（String Interning）机制的文章。Python为了提高内存效率，会自动将一些重复出现的字符串引用到同一内存地址，这与我们之前讨论过的is关键字用于字符串比较时的行为有关。深入理解这一机制，可以帮助开发者避免不必要的内存消耗，并优化程序性能。 此外，近期一篇关于安全编码的文章指出，在进行用户输入验证时，应当避免仅依赖in操作符来判断字符串是否包含敏感词汇，因为这种方式无法防止大小写混淆、边界空格等问题。提倡使用正则表达式或其他专用的安全过滤库，确保字符串匹配严格且全面，以提升应用安全性。 综上所述，掌握Python字符串相等性判断仅仅是字符串处理的基础，实际开发中还需要关注最新的语言特性、内存管理机制以及安全编码规范，才能编写出既高效又健壮的代码。

...编写，同时也易于机器解析和生成。在Web开发中广泛应用于服务器与客户端之间的数据传输，能够存储和表示简单值、数组、对象等多种类型的数据结构，并且可以支持嵌套，便于构建复杂的数据模型。 易语言 , 是一款国产的编程语言及其集成开发环境，具有全可视化编程、跨平台、面向对象等特点，旨在降低编程门槛，提高开发效率。易语言支持丰富的API接口调用和模块导入，使得开发者能够轻松实现各种功能，包括处理JSON数据等高级应用。 JSON模块 , 在易语言环境中，JSON模块是一个专门用于处理JSON数据的内置或第三方库，提供了将JSON字符串解析为易语言可操作的对象（如字典或数组）以及将易语言数据结构转换为JSON字符串的功能。通过调用JSON模块的相关函数，开发者能够便捷地读取、解析、创建和修改JSON数据，以满足Web开发中的数据交换需求。 文件操作模块 , 在易语言编程中，文件操作模块是一组预定义的功能组件，用于执行与文件相关的各种操作，例如打开文件、读取文件内容、写入文件、关闭文件等。结合JSON模块，文件操作模块可用于从JSON文件中读取数据，实现数据持久化存储和加载，进一步增强了易语言对JSON数据流处理的能力。

...能，例如采用了更快的字符串处理函数和改进的内存管理，使得大数据处理更为高效。 值得一提的是，该版本还引入了对JSON数据类型的全面支持，这对于处理复杂的数据结构和API接口变得更为简单。另外，对复制和分区功能的改进，使得在分布式环境中管理大规模数据库变得更加容易。 对于开发者来说，MySQL 8.0的插件式架构允许用户自定义功能，提供更大的灵活性。而对JSON路径查询的支持，使得基于文档的数据查询更加直观。 总的来说，MySQL 8.0是一个值得密切关注的更新，它不仅提升了系统的安全性，而且在性能和功能上都有所突破，是数据库管理员和开发者升级系统的重要参考。随着云计算和大数据的普及，掌握和利用这些新特性将有助于企业在竞争激烈的市场中保持竞争优势。