- 博客(89)
- 资源 (1)
- 收藏
- 关注
RSS订阅原创 row_num引起的数据差异
以card_no分组operate_time倒序排序，会发现最新的两条记录 operate_time相同，status不同【那么不同时间跑有可能这次排序号1的是状态等于1的，下次跑是状态等于-1的】，最终会取状态等于1的且rn等于1的数据。最后然后发现每个ID会有一个update_time，而且最新时间两条不同状态的数据update_time不相同，那么在row_num中将operate_time 修改成update_time更加符合具体业务逻辑，得到的数据也更加准确。拿具体差异的id查看数据。
2023-06-27 16:44:18
177
原创 Streaming黑名单过滤(transform算子的使用)
原始数据：20180808,zs20180808,ls20180808,ww黑名单列表：zsls思路：1、原始日志可以通过Streaming直接读取成一个DStream2、名单通过RDD来模拟一份逻辑实现：1、将DStream转成(黑名单只有名字)(zs,(20180808,zs))(ls,(20180808,ls))(ww,( 20180808,ww))2、然后将...
2019-01-10 16:23:41
694
原创 UpdateStateByKey算子
具体用法参考官网UpdateStateByKey OperationupdateStateByKey操作允许您在使用新信息不断更新时保持任意状态。 要使用它，您必须执行两个步骤。定义状态 - 这个状态可以是任意的数据类型定义状态update函数 - 用这个函数指定如何使用先前状态和新输入流里的新值更新状态对于有状态操作，要不断的把当前和历史的时间切片的RDD累加计算，随着时间的流失，计算...
2019-01-10 16:06:25
731
原创 SparkSQL groupBy的使用
数据格式(消费者ID 消费时间 消费金额)1 12:01 1001 12:02 2001 12:50 1002 12:50 1003 13:01 200需求：统计每个小时,每个用户的消费总额思路步骤：1、id加上时间的小时部分（前两位）作为key2、使用sparkSQl里面的 groupby.agg()方法 groupby(“id”,“time”).agg(sum(“cous...
2019-01-08 11:12:52
9232
原创 Flink MongoDBSink
MongoUtilspackage com.soul.utils;import com.mongodb.MongoClient;import com.mongodb.MongoCredential;import com.mongodb.ServerAddress;import java.util.ArrayList;import java.util.List;/** * @a...
2018-12-24 15:45:03
3605
1
原创 Spark迁移Hive数据到MongoDB(UpdateSave|UpsertSave)
Hive emp表数据hive (soul)> select * from emp;OKemp.empno emp.ename emp.job emp.age emp.deptno7369 SMITH CLERK 24 107499 ALLEN SALESMAN 30 207521 WARD SALESMAN 25 307654 MARTIN SALESMAN 23 1076...
2018-12-20 19:39:58
1671
1
原创 Flink清洗Kafka数据存入MySQL测试
版本信息：Flink Version:1.6.2Kafka Version:0.9.0.0MySQL Version:5.6.21Kafka 消息样例及格式：1.74.103.143 2018-12-20 18:12:00 "GET /class/130.html HTTP/1.1" 404 https://search.yahoo.com/search?p=Flink实战s...
2018-12-20 18:40:38
3796
4
原创 MongoDB的安装及简单使用
本文讲述的是在Mac上如何安装,如果与本文环境不同可以百度自行安装（MongoDB支持Windows、Linux、OS）。一、安装&启动进入 /usr/localcd /usr/local下载sudo curl -O https://fastdl.mongodb.org/osx/mongodb-osx-x86_64-3.4.2.tgz解压sudo tar -zxvf mo...
2018-12-18 21:44:25
314
原创 Mac 安装Hadoop-2.6.0-cdh5.7.0伪分布式
下载Hadoop-2.6.0-cdh5.7.0 单机下载自己需要的版本，我下载的是hadoop-2.6.0-cdh5.7.0一. 解压tar -zxvf hadoop-2.6.0-cdh5.7.0.tar.gz -C xxxxxxxx(自己想放的路径)二 . 配置进入etc/hadoopcore-site.xml<?xml version="1.0" encoding=...
2018-09-24 20:52:30
961
原创 JAR_PARM
JAR_PARM=find ./lib -name *.jar | grep jar | xargs | sed's //,/g'
2018-06-14 20:40:47
242
原创 Hive配置显示当前数据库及表字段
hive-site.xml添加以下参数：<property> <name>hive.cli.print.current.db</name> <value>true</value></property><property> <name>hive.cl
2018-06-08 21:57:41
1848
原创 从输入的SQL参数中获取表名及字段名
工具类package com.isoft.mongo;/* * @Author: Create by Achun *@Time: 2018/6/3 23:20 * */import java.util.ArrayList;import java.util.List;public class SQLUtils { public List<String> g...
2018-06-03 23:42:25
2440
原创 Spark读取mongoDB数据写入Hive普通表和分区表
版本： spark 2.2.0 hive 1.1.0 scala 2.11.8 hadoop-2.6.0-cdh5.7.0 jdk 1.8 MongoDB 3.6.4一 原始数据及Hive表 MongoDB数据格式{ &amp;quot;_id&amp;quot; : ObjectId(&amp;quot;5af65d86222b639e0c2212f3&amp;quot;), &amp
2018-06-03 23:18:15
1644
原创 Spark RDD、DataFrame和DataSet的区别
在spark中，RDD、DataFrame、Dataset是最常用的数据类型，今天谈谈他们的区别！一 、共性 1、RDD、DataFrame、Dataset全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利2、三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算。3、三者都会根据spark的内...
2018-05-18 21:41:02
655
原创 Spark History Server Web UI配置
1.进入spark目录和配置文件[root@hadoop01 ~]# cd /opt/app/spark/conf[root@hadoop01 conf]# cp spark-defaults.conf.template spark-defaults.conf2.创建spark-history的存储日志路径为hdfs上(当然也可以在linux文件系统上)[root@hadoo...
2018-03-12 19:20:40
1502
转载 kafka eagle安装与使用
一、概述 在开发工作当中，在使用 Kafka 集群进行消息的生产和消费时，数据的变动以及topic的管理是我们所关心的内容，这就要求一套行之有效的监控管理体系。常见的kafka监控软件有4种如下所示Kafka Web Conslole ：监控功能较为全面，可以预览消息，监控Offset、Lag等信息，但存在bug，不建议在生产环境中使用。Kafka Manager：偏向Kafka集群...
2018-03-12 19:10:58
2707
原创 Caused by: java.lang.NoSuchMethodError: com.fasterxml.jackson.core.JsonFactory.requiresPropertyOrder
maven中添加 dependency> groupId>com.fasterxml.jackson.coregroupId> artifactId>jackson-coreartifactId> version>2.2.4version> dependency> dependency>
2018-01-31 17:13:58
4013
原创 DN日志--Flume--Kafka
一、Flume收集的DN日志作为Kafka的生产数据,exec-memory-kafka.conf如下a1.sources = r1a1.sinks = k1a1.channels = c1a1.sources.r1.type = com.onlinelog.analysis.ExecSource_JSONa1.sources.r1.command = tail -F /var/lo
2018-01-18 17:27:07
312
原创 DN日志--Flume--HDFS
Flume收集DataNode日志到HDFS之上 一、查看DN日志内容（已经将日志格式改造成了JSON），具体改造方法参考DN日志改造输出为JSON{"time":"2018-01-16 12:07:10,846","logtype":"INFO","loginfo":"org.apache.hadoop.hdfs.server.datanode.DataNode:PacketRespond
2018-01-16 12:15:45
460
原创 Flume编译报错: Failed to execute goal on project flume-ng-morphline-solr-sink: Could not resolve depende
Windows编译Flume源码时报以下错误：[ERROR] Failed to execute goal on project flume-ng-morphline-solr-sink: Could not resolve dependencies for project org.apache.flume.flume-ng-sinks:flume-ng-morphline-solr-sink
2018-01-15 18:12:41
933
原创 修改CDH进程NN、DN日志输出为JSON格式
我们知道SparkSQL是可以直接读取JSON数据的，如果我们要通过Flume采集日志通过Spark处理后进行可视化。那么将日志改造成JSON后在Spark处理阶段就非常方便了。本文主要讲解如何将CDH中HDFS两个进程NN、DN日志改成JSON格式。一、查看DN原始日志格式2018-01-15 11:48:28,916 INFO org.apache.hadoop.hdfs.se
2018-01-15 12:06:42
874
原创 CDH5集群配置lzo
一、安装lzo 1、下载与集群版本相对应的parcel包 parcels下载我CDH版本是5.11.0，所以下载如下三个文件[root@hadoop002 lzo]# lltotal 2440-rw-r--r-- 1 root root 2476671 Jan 12 10:24 GPLEXTRAS-5.11.0-1.cdh5.11.0.p0.30-el6.parcel-rw-
2018-01-12 12:17:35
807
1
原创 Nginx部署
一、Nginx部署依赖几个包，我们首先把依赖包安装好 1、安装PCRE库cd /usr/local/wget http://exim.mirror.fr/pcre/pcre-8.36.tar.gztar -zxvf pcre-8.36.tar.gztcd pcre-8.36make & make install2、安装zlib库cd /usr/local/ tar -zxvf zlib-1.2.8
2017-12-14 18:14:59
241
原创 Elasticsearch配置及head插件安装
ElasticSearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。一、安装配置 ElasticSearch 需要 Java 8 环境，如果不
2017-12-14 16:47:13
471
原创 start-dfs.sh启动HDFS发现datanode没有启动
我们做实验时一般都会搭建一个单节点的集群，有时启动HDFS后发现一直处于Safemode is on，等待很长时间还是没有off后没有耐心就直接重新hdfs namenode -format了，重新启动了发现HDFS的Security is off.Safemode is off.都正常了。此时满心欢喜可以动手做实验，但是悲催的你会发现 hadoop fs -ls 啥的可以使用，但是hadoop
2017-12-11 10:44:08
6942
原创 Azkaban编译及WebServer模式部署
一、下载编译(JAVA需1.8版本) 下载：git clone https://github.com/azkaban/azkaban.git 进入：cd azkaban/gradle/wrapper/目录下 下载：wget https://services.gradle.org/distributions/gradle-4.1-all.zip[root@hadoop001 wrapper]#
2017-12-08 10:06:00
1426
原创 RDDs转DF
一、方式1：反射的方法，但是生产上不建议使用。因为case class只能定义22个字段，有所限制。二、方式2：编程的方式，一般三步走。 1、从原始RDD创建一个RDD[Row]; 2、在步骤1中创建的RDD[Row]中，创建一个与scheam匹配的结构 3、通过SparkSession提供的createDataFrame方法将schema应用于RDD[Row]package com.spar
2017-11-29 12:08:06
729
原创 Scala更改文件目录结构
原文件目录结构hdfs://192.168.187.111:8020/spark/emp/temp/201711112025/d=20171111/h=20/part-r-00000-6ba69620-ba52-4cb1-9ea0-6634ae0e16bc.txt更改后的目录结构hdfs://192.168.187.111:8020/spark/emp/data/d=20171111/h=20/20
2017-11-28 10:49:20
985
原创 Python爬取网站职位信息
# -*- coding = utf-8 -*-import requestsfrom bs4 import BeautifulSoup as bs"""爬取51job里面的工作岗位，公司名称，工作地点，薪资，发布时间"""url = "http://search.51job.com/jobsearch/search_result.php?fromJs=1&jobarea=020000
2017-11-26 11:15:46
499
原创 Spark SQL架构
通过执行计划理解上图spark-sql (default)> explain extended > select > a.key*(4+5), > b.value > from > aa a join aa
2017-11-22 20:54:21
569
原创 Spark SQL 整合Hive的使用
官网地址 点击进入 一、使用Spark SQL访问Hive需要将$HIVE_HOME/conf下的hive-site.xml拷贝到$SPARK_HOME/conf下,如果集群没有安装hive那么就在$SPARK_HOME/conf 下创建一个hive-site.xml，再配置一下可以访问metadata即可二、启动spark访问hive 1、HDFS必须保证启动了 2、如果报错说找不到mysq
2017-11-22 20:44:48
1496
原创 Spark2启动报错
错误如下：[root@hadoop002 conf]# spark-shell --master local[2]Exception in thread "main" java.lang.NoClassDefFoundError: org/slf4j/Logger at java.lang.Class.getDeclaredMethods0(Native Method)
2017-11-16 19:27:07
2893
原创 CDH 自定义添加Spark2服务
官网参考Installing Cloudera Distribution of Apache Spark 2一 、下载SPARK2_ON_YARN-2.2.0.cloudera1.jarSPARK2-2.2.0.cloudera1-1.cdh5.12.0.p0.142354-el6.parcelSPARK2-2.2.0.cloudera1-1.cdh5.12.0.p0.142354-el6.par
2017-11-16 18:49:17
7156
1
原创 CDH Upgrade 5.10.0-->5.11.0
官网Upgrading to CDH 5.6 Using Parcels一、将CM 5.11.0和 CDH 5.11.0的parcels下载好，上传至CM主机。二、进入/val/www/html路径创建5.10.0文件夹，将5.10.0版本的CM和CDH parcel包都移入。三、/val/www/html路径下 1、mkdir cm5/redhat/5/x86_642、tar -zxvf cm5
2017-11-16 17:58:15
535
原创 自定义添加KAFKA服务启动时报错
自定义添加kafka服务参考地址 kafka安装kafka parcels版本如下：（下载地址 download）KAFKA-2.2.0-1.2.2.0.p0.68-el6.parcelKAFKA-2.2.0-1.2.2.0.p0.68-el6.parcel.shamanifest.jsonCDH自定添加kafka服务启动报如下错误 查看stderr:错误日志会发现OOM错误+ grep -
2017-11-16 11:44:19
943
转载 CDH Vcores和DRF
CDH集群调优：内存、Vcores和DRFyurnom2015年08月07日1条评论1,124次浏览吐槽最近“闲”来无事，通过CM把vcores使用情况调出来看了一眼，发现不论集群中有多少个任务在跑，已分配的VCores始终不会超过120。而集群的可用Vcores是360（15台机器×24虚拟核）。这就相当于CPU资源只用到了1/3，作为一个半强迫症患者绝对不能容忍这样的事情发生。分析的过程
2017-11-06 10:58:24
348
转载 CDH优化
一、HDFS dfs.block.size HDFS中的数据block大小，默认是64M，对于较大集群，可以设置为128或264Mdfs.datanode.socket.write.timeout/dfs.socket.timeout 增加dfs.datanode.socket.write.timeout和dfs.socket.timeout两个属性的设置（默认300），比如30000，
2017-11-06 10:56:56
1155
原创 Spark 用户访问量
需求求用户访问量的top5需求分析1、拿到用户 2、访问量求和 2、反转排序再反转一、读取文件val file=sc.textFile("/opt/data/page_views.dat")二、使用tab键分割并拿到第五个字段用户id，然后每个id赋1的操作val user=file.map(x =>(x.split("\t")(5),1))最终效果scala> user.collect.tak
2017-11-05 16:00:05
592
原创 上传代码到Github
这篇文章写得是windows下的使用方法。第一步：创建Github新账户第二步：新建仓库第三部：填写名称，简介（可选），勾选Initialize this repository with a README选项，这是自动创建REAMDE.md文件，省的你再创建。第四步：安装Github shell程序，地址：http://windows.github.com/第五步：打开Git Shell，输入以下命
2017-11-04 21:27:12
279
原创 通过MySQL中元数据信息删除Hive中的表
一、确定Hive中需删除的表名（order_mulit_partition 分区表）hive (default)> desc formatted order_mulit_partition;OKcol_name data_type comment# col_name data_type comment
2017-11-01 17:00:36
2198
空空如也
TA创建的收藏夹 TA关注的收藏夹
TA关注的人