spark如何过滤掉变化时间

我要分析tomcat日志,需要过滤掉不需要的东西
192.168.9.230 - - [24/Apr/2017:19:38:18 +0800] "POST /dvr-gateway/api/genl/paging/device HTTP/1.1" 200 214
比如这一句,我就只想要文件路径和IP地址,该怎么写?用filter可以过滤掉除时间以外的任何数据,但是时间在变,该怎么去过滤?

换句话说我就是在做一个单词统计,统计出每个IP出现的次数和访问文件路径的次数,我能过滤成这样,唯独时间不知道该怎么办了
图片说明

这是我的代码
图片说明

0

1个回答

1
Csdn user default icon
上传中...
上传图片
插入图片
抄袭、复制答案,以达到刷声望分或其他目的的行为,在CSDN问答是严格禁止的,一经发现立刻封号。是时候展现真正的技术了!
其他相关推荐
将SPARK启动和运行时的一些INFO信息过滤掉
我们在每次启动和运行SPARK时,会出现大量的INFO信息,一些WARN和ERROR信息隐藏在其中,这可能会被我们忽略掉,同时对阅读造成了很多不便,所有可以将这些不太重要的信息过滤掉。方法是: 1. 进入$SPARK_HOME/conf目录下,拷贝一个log4j.properties.template,命名为log4j.properties,可以使用下面的语句完成: $ cp log4j.pr
Spark(黑名单过滤)
11.transform以及广告计费日志实时黑名单过滤案例实战 最近看见一个Spark实战,用window做的,挺不错的! transform操作,应用在DStream上时,可以用于执行任意的RDD到RDD的转换操作。 它可以用于实现,DStream API中所有没有提供的操作。比如说,DStream API中,并没有提供 将一个DStream中的么个batch,与一个特定的RDD进行join的
spark sql 过滤空值,过滤某行的空值
data.na.drop() //所有空值 data.na.drop(Seq("col1","col2")过滤第一行,第二行的空值;
使用spark对输入目录的文件进行过滤
使用spark进行文件过滤 在使用spark的很多情形下, 我们需要计算某个目录的数据. 但这个文件夹下面并不是所有的文件都是我们想要计算的 比如 : 对于某一天的数据,我们只想计算其中的几个小时,这个时候就需要把剩下的数据过滤掉 更坏的一种情形 : 对于那些正在copy(还没有完成),或者是.tmp临时文件, 程序在读取的过程中,文件发生变化已经复制完成或者被删除,都会导致程序出错而停掉 ...
Spark Streaming实时处理本地数据流
每隔20s监听本地文件夹“/home/hduser/Streamingtext”下新生成的文本文件,对新文件中的各单词个数进行统计/* Streamingtext下操作文件应注意以下几点: 1.监控目录下的文件应该具有统一的数据格式,避免在内部解析时报错。 2.文件必须是在监控目录下创建,可以通过原子性的移动或重命名操作,放入目录。 3.一旦移入目录,文件就不能再修改了,如果文件是持续写入的话,新的
scala flatMap为何能过滤掉Option中的None
import scala.collection.mutable._ def indexes(a: Array[String], m: Map[String,Int]) = a.flatMap(m.get(_)) def indexes1(a: Array[String], m: Map[String,Int]) = a.map(m.get(_)) val am = Array("Tom", "Fr
Spark Streaming之:Flume监控目录下文件内容变化,然后Spark Streaming实时监听Flume,然后从其上拉取数据,并计算出结果
1、安装flume 2、到Spark-Streaming官网下载poll方式的Sink 3、将sink放入到flume的lib包里面 4、先启动flume(多个),然后在启动Streaming程序下载spark-flume http://spark.apache.org/documentation.html 到Spark-1.6.2中 http://spark.apache.org/do
2,StructuredStreaming的事件时间和窗口操作
推荐阅读:1,StructuredStreaming简介使用Structured Streaming基于事件时间的滑动窗口的聚合操作是很简单的,很像分组聚合。在一个分组聚...
Spark 1.4新特性概述
经过4个RC版本,Spark 1.4最终还是赶在Spark Summit前发布了,本文简单谈下本版本中那些非常重要的新feature和improvement. 正式引入SparkR   Spark 1.4正式引入了SparkR(可以参见本博客的《Spark官方正式宣布支持SparkR(R on Spark)》介绍),它是一个R API,SparkR是基于Spar
java,spark实现黑名单过滤
/** * java,spark实现黑名单过滤 */ public class BlackListFilter { public static void main(String[] args){ SparkConf conf = new SparkConf().setAppName("Simple Application").setMaster("local[2]")...
如何解决spark开发中遇到需要去掉文件前几行数据的问题
转载请务必注明原创地址为:https://dongkelun.com/2018/05/27/sparkDelFirstNLines/ 前言 我碰到的问题是这样的,我需要读取压缩文件里的数据存到hive表里,压缩文件解压之后是一个txt,这个txt里前几行的数据是垃圾数据,而这个txt文件太大,txt是直接打不开的,所以不能手动打开删除前几行数据,而这个文件是业务人员从别人那拿到的所以也不能改,...
scala程序写Spark程序实现黑名单过滤
package com.bigdata import org.apache.spark.rdd.RDD import org.apache.spark.{SparkContext, SparkConf} /**  *  */ object BlackWhite {   def main(args: Array[String]) {     createContext
spark ML ALS实现电影推荐
package zqr.com; 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 o...
spark笔记之通过spark实现点击流日志分析案例
1. 访问的pv package cn.itcast import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object PV { def main(args: Array[String]): Unit = { //todo:创建sparkconf,设置ap...
利用Hadoop和Spark处理用户心跳周期数据
数据源:可穿戴设备的实时数据分析。1.txt记录的是某一个用户的心跳周期数据,每一个数值表示一次心跳的周期,单位是秒。例如,0.8表示用户当时的心跳间隙是0.8秒。心跳间期按照顺序存储
SparkSQL之查询,过滤
def main(args: Array[String]): Unit = { val conf = new SparkConf().setMaster("local").setAppName("ScalaSparkDataFrameOps") val sc = new SparkContext(conf) val sqlContext = new SQLContext(sc)
Spark Streaming处理速度滞后于读取速度测试
Java代码   包 spark.examples.streaming      导入 org.apache.spark.SparkConf   import  org.apache.spark.streaming.StreamingContext._   导入 org.apache.spark.streaming._      对象NetCatStreamingWord
Spark Streaming 流计算优化记录(4)-时间都去哪儿了,关于调度与空转
关于Spark Streaming优化的各种叽里呱啦记录, 这一次是从跑不动, 一直优化到能每秒解决6万条输入消息以及3G数据的Inner Join. 大数据,流计算,spark,kafka,hadoop
Spark GC调优笔记
Spark GC调优笔记传统GCParalledl GC:高吞吐量,离线分析 CMS GC:低延迟,实时响应 Java堆内存划分为两个区域:Young和Old,Young存储短期对象,Old存储长期对象。 Young代进一步划分为三个区域:Eden,Survivor1,Survivor2 GC调优的目标只有长期对象存储在老年代,新生代有充分的大小短期对象。调优方法 如果老年代接近满,降低spark
“戏”说Spark-Spark核心-RDD转换操作算子详解(一)
对于RDD可以有两种计算方式: 转换(返回值还是一个RDD)---懒执行 操作(返回值不是一个RDD)---立即执行 转换(Transformations) (如:map, filter, groupBy, join等),Transformations操作是Lazy的,也就是说从一个RDD转换生成另一个RDD的操作不是马上执行,Spark在遇到Transformations操作时只会记录需要这样的操作,并不会去执行,需要等到有Actions操作的时候才会真正启动计算过程进行计算。 操作(Actions) (
Learning Spark笔记4-常见的转换与动作
基本RDDs 不管数据如何,我们先将介绍一下我们可以对RDDs做哪些转换和操作 1.元素转换 这里有两个常用的转换map()和filter()。 map()转换会应用到RDD中的每个元素上,函数的结果是RDD,该RDD中的每个元素都有一个新值。 filter转换返回的结果是RDD,该RDD只包含通过filter()的元素。 我们可以使用map()
Spark项目练习(计算用户停留时间最长的两个小区)
项目说明:附件为要计算数据的demo。点击打开链接 其中bs_log文件夹数据格式为(手机号,时间戳,基站ID,连接状态(“1”为连接,“0”为断开)) lac_info.txt 文件数据格式为(基站ID,经度,纬度,信号辐射类型) 程序思路: 1, 先根据"手机号,基站ID"构成一个元祖,做为唯一标识, 和时间戳构成新的数据结构->(手机号, 站点, 时间戳) 2、(手机号,基站ID)
理解SparkSteaming窗口函数操作window()
需求场景:      一些业务场景,例如网站记录,每隔1个小时计算最近两个小时的pv量,还有一种业务场景的话先在内存中做累加再更新到redis中做累加,比如说每隔5秒统计最近5秒的数据的总和,再刷到redis中做累加,因为频繁操作redis的话会存在问题。   重要参数: 1.批处理间隔 2.窗口间隔 3.滑动时间间隔   原理介绍:        在Spark Streamin...
Spark---算子调优之filter过后使用coalesce减少分区数量
默认情况下,经过了这种filter之后,RDD中的每个partition的数据量,可能都不太一样了。(原本每个partition的数据量可能是差不多的) 问题: 1、每个partition数据量变少了,但是在后面进行处理的时候,还是要跟partition数量一样数量的task,来进行处理;有点浪费task计算资源。 2、每个partition的数据量不一样,会导致后面的每个tas
Spark-SQL常用的时间处理函数
Spark-SQL常用的时间处理函数 Spark-SQL函数对大小写不敏感 姊妹篇:Mysql常用时间处理函数 year select year('2018-03-03 20:30:34'); // 2018 month select month('2018-03-03 20:30:34'); // 3 weekOfYear select weekOfYear(...
spark1.x升级到spark2.x以及1.x和2.x的版本兼容
1. spark 1.x 升级到spark 2.x 对于普通的spark来说,变动不大 : 举一个最简单的实例: spark1.x public static JavaRDD<String> workJob(JavaRDD<String> spark1Rdd) { JavaPairRDD<String, Integer> t...
Spark成长之路(2)-RDD中分区依赖系统
参考文章: 深入理解Spark RDD抽象模型和编写RDD函数 RDD依赖 spark调度系列 偏函数 简介 依赖关系图 依赖概念类 窄依赖类 OneToOneDependency RangeDependency PruneDependency 宽依赖类 图解 ShuffleDependency简介RDD之间的依赖关系大致分为两类:窄依赖和宽依赖。 借用参考文章的解释: 窄
spark容错机制
集群容错机制Master异常退出后重启:不影响退出之前已经提交的application的运行,但是在退出期间exector的资源释放,异常退出重新调度等功能会受到影响;新的appliaction无法提交;重新启动后原来的已经创建的应用信息和driver信息不会重新上报到master,原有的worker依然会通过heartbeat心跳信息上报,worker检测到master的退出,会重新发出注册的请...
spark-hbase数据操作心得
通过spark程序读写hbase的方法百度上太多了,这里我就不一一列举,这里我要分享的是我在开发spark程序对hbase读写时喜欢使用的方法,水平有限,还望指点。我的hbase表结构是简单的rowkey + cfamily:”INFO” + value首先,定义几个方法/** * 扫描全表,返回数据集 * @param connection: Connection * @param tableNa
spark sql 获取系统时间,计算时间后结果浮点型转整型
spark sql 获取当前系统时间  org.apache.spark.sql.AnalysisException: Undefined function: 'getdate'
随时间变化提示.rar
随时间变化提示.rar随时间变化提示.rar随时间变化提示.rar随时间变化提示.rar随时间变化提示.rar随时间变化提示.rar随时间变化提示.rar随时间变化提示.rar
Spark 应用程序调优
Spark 应用程序调优
Spark2.1.0官方文档
Spark 概述Apache Spark是一个快速和通用的集群计算系统。它提供Java,scala,Python、R语言的APIs,以及支持一般执行图形的优化引擎。 它还支持一组丰富的高级工具,包括用于SQL和结构化数据处理的Spark SQL,用于机器学习的MLlib,用于图形处理的GraphX和Spark Streaming。
一款可以限定选择区域的日期控件
这个开源控件可以兼容到SDK8版本,可以自定义显示的年月日,以及时间范围,如图     如果我们只想显示两个月的日期选择区间: final Calendar month = Calendar.getInstance(); month.add(Calendar.MONTH, 1); 然后初始化时候可以传进去一些参数 1 final CalendarPickerView
hive 和spark sql 中日期相关使用技巧(一)
hive 和spark sql 中日期相关使用技巧 常用日期/时间相关字段、类型、及示例: 类型包括:BIGINT,TIMESTAMP和STRING 下面是这几种类型之间相互转换的方法: BIGINT转TIMESTAMP:from_unixtime(time/1000) TIMESTAMP转BIGINT:unix_timestamp(time)*1000 BIGINT转STRING(y...
Spark性能调优-性能监控、参数调整
以下是整理的Spark中的一些配置参数,官方文档请参考Spark Configuration。 Spark提供三个位置用来配置系统: Spark属性:控制大部分的应用程序参数,可以用SparkConf对象或者Java系统属性设置环境变量:可以通过每个节点的 conf/spark-env.sh脚本设置。例如IP地址、端口等信息日志配置:可以通过log4j.properties配置
使用Spark ALS实现协同过滤
转自:http://blog.javachen.com/2015/06/01/how-to-implement-collaborative-filtering-using-spark-als.html 本文主要记录最近一段时间学习和实现Spark MLlib中的协同过滤的一些总结,希望对大家熟悉Spark ALS算法有所帮助。 更新: 【2016.06.12】Spar
Spark缓存清理机制
unpersist http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html#unpersist Dematerializes the RDD (i.e. Erases all data items from hard-disk and memory). However, the RDD object remains
Spark Streaming性能调优详解
Spark Streaming提供了高效便捷的流式处理模式,但是在有些场景下,使用默认的配置达不到最优,甚至无法实时处理来自外部的数据,这时候我们就需要对默认的配置进行相关的修改。由于现实中场景和数据量不一样,所以我们无法设置一些通用的配置(要不然Spark Streaming开发者就不会弄那么多参数,直接写死不得了),我们需要根据数据量,场景的不同设置不一样的配置,这里只是给出建议,这些调优不一
Spark本地化策略导致的任务执行时间异常
本地化执行时间异常 这两天遇到一个很奇怪的问题,利用receiver和direct两种方式来消费kafka中的数据,官网说direct形式的效率更高但是实际效果却不是很好没有那么快。 这是因为经过查看运行时间表发现,每个批次都等待了3秒的时间。因此联想到spark.locally.wait的默认等待时间是三秒因此可能是为了等待数据的本地化导致的时间延长,因此把时间调低来解决。本地化策略是为了减...
文章热词 Spark Spark培训 Spark课程 Spark视频教程 Spark学习
相关热词 c++ mlib spark 调用 c++如何比较时间 android8.0的变化 spark教程+python spark教程python