- 博客(104)
- 收藏
- 关注
RSS订阅
原创 错误总结
1:导入APP端埋点日志数据到ods的时候,不小心又导入了wx小程序端埋点日志,还好刚刚创建的ods我不应该删除这个表,应该用overwrite,直接覆盖event_app_log表load data local inpath “/opt/data/inlogdata/event_log” overwrite into table event_app_log; 2:创建公共表的时候要用外部表external...
2020-09-01 10:41:47
68
原创 HDFS数据迁移
跨集群传输数据：(由nameservice2集群传入nameservice1集群)这里传输的时候目录要以/结尾，不行就在/添加*hadoop distcp -skipcrccheck hdfs://nameservice2/user/hive/warehouse/ods_yspay_trade.db/ods_orderbill/ hdfs://nameservice1/user/hive/warehouse/temp.db/orderbill_test/1传入之后hive查询不到数据：如果添加
2021-03-31 11:01:45
347
原创 2021-03-18
select * from dwd.settlement_user_pay where （p_batch_code != '20210130_TCBXP(JK-HX-SF)' ） or(a_system !='jk' and b_system !='hx' and p_batch_code = '20210130_TCBXP(JK-HX-SF)')（p_batch_code != '20210130_TCBXP(JK-HX-SF)' 取的是settleme...
2021-03-18 11:14:08
191
原创 instr
dwd.settlement_user_pay_month a LEFT JOIN dwd.sf_user_pay_month b ON IF ( INSTR(b.policyno, '_') > 0, SUBSTRING( b.policyno, ...
2020-12-15 11:11:15
217
原创 数据质量管理
一丶数据治理管理的意义对数据仓库建设具有指导意义，是数据仓库建设成功与否的重要衡量指标。能帮助业务系统找到一些不易发现的问题，从侧面反映业务系统的可靠性。有利于提升用户对数据仓库的信任度。二丶元数据管理层面1丶DMP,元数据系统和业务库,hive表结构要保持完全一致,不能缺少字段(一)丶数据层面1丶重复校验对于表的主键以及唯一约束等需要进行重复校验如:单个字段校验,联合字段校验等2丶格式校验(1) 非空校验（’\Null’ , ‘’, Null）(2) 字段长度校验(>、&
2020-12-03 09:33:05
309
原创 hive时间sql
Hive 中，可以用String、Date和Timestamp表示日期时间，String 用 yyyy-MM-dd 的形式表示，Date 用 yyyy-MM-dd 的形式表示，Timestamp 用 yyyy-MM-dd hh:mm:ss 的形式表示。这三种数据类型在使用细节上，有一些需要注意的点：在这里插入图片描述在这里插入图片描述Join比较在两表Join时，会涉及到字段的比较，此时应注意：如第一张图所示，如果时间信息中不包含时分秒，String 与 Date、Timestamp 表达的时.
2020-09-12 23:34:30
607
原创 拉链表
拉链表思路2号拉链1,ae,2020-08-25,2020-08-311,aa,2020-09-01,9999-12-312,bb,2020-09-01,2020-09-012,ba,2020-09-02,9999-12-313号增量1,ab3,cc1:拉链表是将数据更加简化,让节约数仓储存空间让数据更加利于计算计算逻辑,如 1 号用户ae假设为加入购物车,.那么当它加入购物车的时候就给出2020-08-25 到9999-12-31那么如果它有下一次操作,就以当...
2020-09-11 23:15:47
266
原创 业务系统与sqoop
1:业务系统业务域的数据来自业务系统的数据库通过sqoop（或datax）抽取到数仓的ods层在ods层对有需要的表进行增量合并，字段选择，反范式话，形成dwd明细层表在明细层基础上，进行各类主题的数据统计、分析课程中，主要分析的主题有：交易域分析 营销域分析 会员域分析2:sqoop1:sqoop是apache旗下的工具主要用于服务器之间传送数据的工具主要核心功能为导入 导出导入:从mysql数据库中导入到hdfs导出:从 Hadoop 的文件系统中.
2020-09-11 22:51:56
158
原创 漏斗sql
regexp_extract函数regexp_extract(str, regexp[, idx])参数解释:其中：str是被解析的字符串或字段名regexp 是正则表达式idx是返回结果 取表达式的哪一部分 默认值为1。0表示把整个正则表达式对应的结果全部返回select regexp_extract('hitdecisiondlist','(i)(.*?)(d)',0) ; 返回的是i-d直接的数+------+| _c0 |+---...
2020-09-10 20:31:54
309
原创 新老用户留存
guid和first_dt,rng_start很好解决,rng_end需要判断.昨天不在今天在,那么保留老记录,新添加一行 ----新添加这一行是09 05 -9999-12-21昨天在今天也在 那么嗨是保留记录昨天不在今天也不在保留记录 --保留原纪录）昨天在,今天不在 =T-1新用户 将二表full j...
2020-09-09 22:01:08
228
原创 first_value和last_value
SQL中分析函数first_value(),last_value，sum() over(partition by…)详解首先，生成有一张原始表score，s_id表示学生id，c_id表示课程id，s_core表示最后课程的成绩，表结构数据如下所示：首先，按照课程c_id进行分组，按照成绩s_core进行排序select * ,row_number() over (partition by c_id order by s_score)rank_showfrom score1运行结果：
2020-09-07 19:40:08
2920
2
原创 数仓脚本
#!/bin/bash########################################### # @desc: app端埋点日志文件导入ods层表# @target: ods.event_app_log##############################HIVE_HOME=/opt/apps/hive-3.1.2/DT=`date -d'-1 day' +%Y-%m-%d`if [ $1 ]thenDT=$1fi${HIVE_H...
2020-09-07 18:24:13
664
原创 上传jar包到集群
我们的项目是父子工程,jar包都是互相依赖的所以单独打某个jar是不行的,我们将xml加入到想运行的子工程内<build> <plugins> <!-- 把依赖jar中的用到的类，提取到自己的jar中 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifact
2020-09-07 18:21:58
392
原创 hivesql
语法：分析函数over(partition by分组列 order by排序列rows between 开始位置 and 结束位置)常用分析函数： 聚合类 avg()、sum()、max()、min() 排名类 row_number() 按照值排序时产生一个自增编号，不会重复 rank() 按照值排序时产生一个自增编号，值相等时会重复，会产生空位 dense_rank() 按照值排序时产生一个自增编号，值相等时会重复，不会产生空位 其他类 lag(列名,往...
2020-09-06 22:34:06
105
原创 雪花模型和星型模型和三范式
星型模型和雪花模型在多维分析的商业智能解决方案中，根据事实表和维度表的关系，又可将常见的模型分为星型模型和雪花型模型。在设计逻辑型数据的模型的时候，就应考虑数据是按照星型模型还是雪花型模型进行组织。星型模型当所有维表都直接连接到“事实表”上时，整个图解就像星星一样，故将该模型称为星型模型。星型模型星型架构是一种非正规化的结构，多维数据集的每一个维度都直接与事实表相连接，不存在渐变维度，所以数据有一定的冗余，如在地域维度表中，存在国家A省B的城市C以及国家A省B的城市D两条记...
2020-09-05 22:17:03
3334
1
原创 序列化浅谈kyro
序列化经过网络传输,shuffle,将对象转化为二进制.最终落写磁盘需要反序列化解码二进制jdk序列化传入类和对象还有对象内的属性-- 类,和对象==类的元信息因为如果只传入对象,都是二进制不知道从哪分割,如果类还继承了其他类那么也需要将它的父类也要序列化缺点:不仅序列化了属性数据值,还序列化了父类和类的元信息写入磁盘io很大,磁盘占用字节大hadoop序列化交给用户自己定义setmapoutputKeyClass(Text.class) 输入的类反射,下游task知道上游传..
2020-09-05 20:39:40
1331
原创 idmaping代码实现
object IdMappingGen { def main(args: Array[String]): Unit = { Logger.getLogger("org").setLevel(Level.WARN) val spark = SparkSession.builder().appName("IDMAPPING映射字典生成").master("local[*]").getOrCreate() import spark.implicits._ import or.
2020-09-04 22:56:31
301
原创 数仓数据清理
今天的需求是清洗数据,给用户添加省市区,添加新老用户,添加idMaping(用户标识)1清洗过滤数据1.1过滤掉日志account及deceid全为空的记录"!(isBlank(deviceid) and isBlank(account))")1.2过滤日志中缺少关键字端properties/eventid/sessionid 缺任何一个都不行"properties is not null and !isBlank(eventid) and !isBlank(sessionid)"1.3过.
2020-09-04 22:40:25
1065
原创 布隆过滤器
布隆过滤器,不用将数据存入 byte数组里, 而是存入bit (比特)里通过hashcode%字符串,将hashcode结果映射到bit里,存入为1 不存在为0有可能hash碰撞,本来不在,误判为在, 不可能是不在误判为不在(hbase在底层是通过布隆过滤器,那么如果查找rowkey本来不在确判定在,那么也没有上面影响只是浪费时间查找,不会存在,本来rowkey在这个hfile里,但是出现0不在的情况)改进,用多种hash算法,对同一个事物得到多个hash值映射到bit数组的多个位置 或者增加b..
2020-09-04 20:10:03
110
原创 Union和UnionAll的区别
假设我们有一个表Student，包括以下字段与数据：drop table student; create table student ( id int primary key, name nvarchar2(50) not null, score number not null ); insert into student values(1,'Aaron',78); insert into student values(2,.
2020-09-01 22:16:54
426
原创 spark内存的前世今生
1.前言在执行Spark的应用程序时，Spark集群会启动Driver和Executor两种JVM进程，前者为主控进程，负责创建Spark上下文，提交Spark作业（Job），并将作业转化为计算任务（Task），在各个Executor进程间协调任务的调度，后者负责在工作节点上执行具体的计算任务，并将结果返回给Driver，同时为需要持久化的RDD提供存储功能。由于Driver的内存管理相对来说较为简单，本文主要对Executor的内存管理进行分析，下文中的Spark内存均特指Executor的内存。
2020-09-01 21:35:55
137
原创 spark在yarn上的二种运行模式和Spark StandAlone模式
RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。1:spark的StandAlone模式1.1 spark的client模式特点:spark的client模式的特点是Driver端在SparkSubmit里1:启动Master,master负责集群资源管理 ,worker负责监控自己的cpu,内存消息向master汇报2:在client中启动D
2020-09-01 21:03:42
309
原创 日志埋点说明
1:日志埋点是业务系统开发端的工作,数据开发人员仅需要提出数据埋点需求埋点可以植入到业务系统后端和前端埋点生成的日志统一为JSON格式,各个终端渠道的埋点日志，都由公共属性字段，和事件属性字段组成；不同终端渠道，公共属性字段略有不同； 事件属性则根据事件类型，灵活多样；一条日志一般500字节作用,一条日志有多少字段40,50个,日志有事件详情,不同事件的字段是不同的如{用户账户 appid,公司内部开发用于的groupid ...
2020-09-01 10:58:42
2618
原创 Kafka会丢失数据吗?
生产者首先发送消息是向leader发送,当ack通信机制=0的时候(leader立刻给生产者回信,已经收到消息了)leader收到消息了会收集到内存内,相当于记录了偏移量,生产者消息还没有发送完,leader就挂了,或者,leader通过IO发送给follower,有时间差这个时候就会造成消息的丢失.如何解决:ack=1 : 有一个副本持久化,也就是leader完全接受到信息持久化,然后在回复生产者收到消息了 -ack=-1 所有的 isr副本全部同步完成相对于效率来说 ac...
2020-09-01 10:30:34
462
原创 flume从kafka采集到hdfs配置文件编写
-- fume的采集配置a1.sources = r1a1.channels = c1a1.sinks = k1# source configa1.sources.r1.channels = c1a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource --从kafka收集信息a1.sources.r1.kafka.bootstrap.servers = doitedu01:9092,doitedu02:9092,doit.
2020-09-01 10:08:07
296
原创 flume浅谈
1.1 什么是flume?Flume是apache旗下的数据采集工具,主要目的是收集日志,事件等资源官网简介 Flume是一种分布式、可靠和可用的服务，用于高效地收集、聚合和移动大量日志数据。它有一个简单而灵活的基于流数据流的体系结构。它具有可调的可靠性机制和许多故障转移和恢复机制，具有健壮性和容错性。它使用一个简单的可扩展数据模型，允许在线分析应用程序。它可以将从各种各样的数据源上采集数据,并传输到 Hadoop生态圈(hbase,hive,hdfs,kafka)1.2 Flume特.
2020-08-30 21:53:48
126
原创 linux同步时间详细步骤
1:下载ntpdateyum install -y ntpdate2:ntpdate ntp1.aliyun.com同步时间
2020-08-29 20:18:30
105
原创 linux每次启动重置ip解决方案
跟系统自带的NetworkManager这个管理套件有关系，关掉就可以解决。运行systemctl stop NetworkManagersystemctl disable NetworkManager然后重启网络systemctl start network.service重启机器reboot
2020-08-29 15:47:26
433
原创 SparkStreaming聚合过的（事物）保证Exactly-Once
package com._51doit.spark.day15import java.sql.{Connection, PreparedStatement}import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.kafka.common.TopicPartitionimport org.apache.kafka.common.serialization.StringDeserializerimport .
2020-08-28 14:23:17
149
原创 sparkStreaming消费着Kafka将数据写入mysql中
/** * @author HANGGE 2020/3/10 21:58\消费kafka中的数据 */object Demo1 { Logger.getLogger("org").setLevel(Level.ERROR) System.setProperty("HADOOP_USER_NAME", "root") def main(args: Array[String]): Unit = { //定义状态更新函数 val updateFunc = (values.
2020-08-27 23:14:42
785
原创 spark二表关联写入mysql中
package com._51doit.spark.persimport java.sql.{Connection, Date, DriverManager, PreparedStatement, SQLException}import com.alibaba.fastjson.{JSON, JSONException}import org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}object.
2020-08-27 22:49:49
119
原创 spark连接kafka自己定义
package cn._51doit.spark.day13import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.kafka.common.serialization.StringDeserializerimport org.apache.spark.SparkConfimport org.apache.spark.streaming.dstream.{DStream, InputDStream}imp.
2020-08-27 22:36:36
100
原创 Spark连接kafka官方直连方式
kafka默认提交偏移量,消费者会将偏移量默认写到特殊的topic,偏移量更新的时候,数据已经处理,但是还没有更新偏移量,再次重新启动,会重复提交偏移量,控制不够精准,无法保证数据的一致性---所有我们不让消费者自动提交偏移量 :"enable.auto.commit" -> (false: java.lang.Boolean)偏移量是描述信息,偏移量在Driver端生成在Driver获取...
2020-08-27 22:35:20
562
原创 clickhouse和namenode端口冲突
在linux中启动clickhouse后,namenode不能启动,查资料发现是端口被占用,于是就找到以下解决办法: netstat -a -t --numeric-ports -p 列举所有端口号及在使用的进程发现9000端口被clickhouse占用 tcp6 0 0 [::]:9000 [::]:* LISTEN 1618/clickhouse-ser 修改clickhouse的配置...
2020-08-25 19:38:02
667
原创 KafuKa:代码实现
1:topic ,分区,groupID影响着分区当更换了groupID,以前没有记录过就从头开始消费(earliest),消费完毕,就会记录下来,在读就不会重复读取,//earliest：从头开始 --from-beginning //latest:从消费者启动之后当是latest,即使更换了groupID,也不会读取数据,消费者第一次启动才会读取数据2://props.setProperty("enable.auto.commit", "false") // 不让kafka...
2020-08-23 11:26:51
383
原创 redis Hash类型 Map＜String, Map＜String, String＞＞
package day01import java.utilimport redis.clients.jedis.Jedisobject HashValueDemo1 { /** * Author: HashValueDemo1,相当于一个大key包着一个map集合== (key,(key,value)) * Date: 2020/8/22 * Description: */ def main(args: Array[String]): Unit .
2020-08-22 21:43:45
575
原创 redis: SortedSet（zset） Map＜String, TreeSet＞
package day01import java.utilimport redis.clients.jedis.Jedisobject ZsetValueDemo { /** * Author: SortedSet（zset） Map<String, TreeSet> 可排序的,对于求topN很适合 * Date: 2020/8/22 * Description: */ def main(args: Array[String]): .
2020-08-22 21:42:47
374
原创 redis Set Map＜String, HasSet＞
package day01import java.utilimport redis.clients.jedis.Jedisobject SetValueDemo1 { /** * Author: JEDIS的set类型 * Date: 2020/8/22 * Description: */ def main(args: Array[String]): Unit = { //输入linux名,和6379 val jedis = ne.
2020-08-22 21:42:10
149
原创 redis的List Map＜String, List＞
package day01import java.utilimport redis.clients.jedis.Jedisobject ListValueDemo1 { /** * Author: List的数据库,传入一个key,后面是list * Date: 2020/8/22 * Description: */ def main(args: Array[String]): Unit = { import scala.collecti.
2020-08-22 21:39:36
1096
1
空空如也
空空如也
TA创建的收藏夹 TA关注的收藏夹
TA关注的人