小白鸽-CSDN博客

原创实时数据开发

笔者并没有专业的实时数仓的开发经验，只是从别人那和网上资料综合整理而来，仅供参考。实时数据开发步骤：1、需求分析；2、确定Source、Sink、Dim；3、任务创建；4、任务开发和修改；5、参数与资源配置；6、任务发布；7、任务运维。

2024-04-12 19:13:13 280

原创数据治理(二)-基线治理

Mapjoin/Auto Mapjoin参数不合理。Map/Join/Reduce参数不合理。ServiceMode参数不合理。小fuxi instance治理。超大merge任务治理。logview异常耗时处理。被高频访问的adm视图治理。D2/DG互通链路治理。D2/DG的DQC治理。高频预警/破线治理。性能差的自定义函数治理。Owner离职/转岗。

2024-04-10 22:23:08 341

原创用户增长方法论

用户增长的精髓是一套体系和方法，它以数据为驱动，以实验的方式，系统的在用户生命周期的各个阶段(包含用户获得、激活、留存、推荐、变现、回流等)，寻找当下性价比最高的机会，在具体执行上横跨市场、产品、工程、设计、数据等团队，通过快速迭代实验的方法达到目标。

2024-04-04 22:28:47 238

原创数据治理-平台治理

1.待治理成本2.浪费量：可以优化计算或者存储3.累计浪费量：发现时距离今天数*当日浪费量4.节省成本：完成治理成本5.白名单管理成本。

2024-03-31 23:04:50 277

经常变化口径，并需要回刷数据的业务中间层，可设计基础中间层+视图业务中间层的方式解决。：为了查询方便，事实表会冗余多个字段，若为了冗余字段导致调度启动太晚，可以拆分快慢表。：依赖多个主表，且每个主表时间差异太大，可拆二级分区，二级前置节点。：所有表，必须考虑是否设置逻辑主键，理论上所有表都存在逻辑主键。：主要针对事实表，大约100E条以上的事实表，设计增量。1.数据架构域与在线应用架构域大部分能够实现映射。：设计维度、度量、中间层时，保持口径一致。：常见维度属性，冗余到事实表、维表。

2024-03-30 09:36:29 338

原创【数据分层方法论】初稿

一般标准数据架构有(ODS、DWD、DWS、DIM、ADS)

2024-03-29 11:11:54 284

原创 Spark-sql array<struct>中 struct字段截取（spark2.4版本以上）

原数据{"date":"20200512","from":"银川","requestTime":"14:12:38","sceneCode":"010035","seats":[{"flag":0,"oldPrice":0,"price":"70","seatFlag":false,"seatName":"二等座","seats":21},{"flag":0,"oldPrice":0,"price":"112","seatFlag":false,"seatName":"一等座","seats":21},{

2020-05-12 17:59:23 2343

原创从IO到NIO笔记

最近组里小伙伴做了次分享，感觉挺好的，虽然我不是后端开发，拿出来整理整理，做做笔记。基础概念内核是操作系统的核心软件。它独立于普通的应用程序，可以访问受保护的内存空间，也有访问底层硬件设备的所有权限，为应用程序提供对计算机硬件的安全访问服务。为了保证内核安全，避免用户进程直接操作内核，操作系统将内存寻址空间划分为两部分：（1）内核空间Kernel-space（2）用户空间：供内核程...

2020-04-29 11:55:58 382

原创 Flink数据统计UV、PV统计(三种写法)

做了那么多次flink统计，发现我居然没写过uv，pv统计（因为uv,pv实时统计,公共平台帮做了）。 public static final DateTimeFormatter TIME_FORMAT_YYYY_MM_DD_HHMMSS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); ... Properties prop...

2020-04-10 11:25:33 9173 17

原创 Java时间转换小技巧

SimpleDateFormat有多线程问题，尽量使用DateTimeFormatter当前时间加+1，返回结果是2020-03-12； LocalDate localDate = LocalDate.now(); localDate = localDate.plusDays(1); System.out.println(localDate);...

2020-03-11 16:34:56 218

原创 Flink MapState的ConcurrentModificationException问题

直接上代码import com.alibaba.fastjson.JSON;import com.tc.flink.analysis.label.bean.output.ItemIdWithAction;import org.apache.flink.api.common.functions.RichMapFunction;import org.apache.flink.api.commo...

2020-01-13 15:46:32 1556 1

原创面试题ArrayBlockingQueue优化问题

同事讨论一道面试题，是ArrayBlockingQueue里代码问题是326行是多余的吗？所在意义是什么。因为是jdk源码，肯定有意义的。也有答案第二个原因，为了安全，我们很容易一致同意。问题在于原因一，为什么变成读方法里局部变量是一条指令，读类的成员变量是两条指令、所以性能优化。上网找了一些材料、解读。...

2019-12-12 14:41:23 549

原创 Flink 自定义Trigger

需求，滑动窗口统计，keyby下过来一条就触发窗口统计，如果没消息过来，按30s触发一次窗口。只能自定义Trigger直接上代码import com.alibaba.fastjson.JSON;import com.alibaba.fastjson.JSONObject;import com.tc.flink.conf.KafkaConfig;import org.apache.fli...

2019-11-06 16:34:10 1330 3

原创 ElasticSearch相关概念

最近整理一些es面试的相关内容。大部分来自于网上信息整理。基础概念节点（Node）：物理概念，一个运行的Elasticearch实例，一般是一台机器上的一个进程。索引（Index）: 逻辑概念，包括配置信息mapping和倒排正排数据文件，一个索引的数据文件可能会分布于一台机器，也有可能分布于多台机器。索引的另外一层意思是倒排索引文件。分片（Shard）：为了...

2019-09-09 15:25:18 214

原创 Flink打包编译问题

在打包编译1.9 flink时候，出现这个问题。[ERROR] Failed to execute goal on project flink-avro-confluent-registry: Could not resolve dependencies for project org.apache.flink:flink-avro-confluent-registry:jar:1.9-SNAP...

2019-08-14 14:35:15 3003

原创 Flink排查watermark坑

最近有这样需求，消息中间件可能有延迟消息，比如延迟5分钟后数据，可以摒弃。参考TimeLagWatermarkGenerator DataStream<Tuple2<String,Integer>> filterStream= stream.filter(s -> s != null && !s.trim().equals("")) ...

2019-08-08 16:53:54 987 1

原创 Flink 再谈State Backends

最近做数据统计时候，在做大状态数据管理时候，发现问题。因为大状态，我没开checkpoint。申请了6个容器资源，但半小时内就出现slot container丢失，重启。每隔半小时重启，但开了（ck）checkpoint，采用RocksDBS发现很少出先slot丢失重启。一直不明白，开始以为ck会消耗性能，没开CK为什么会经常重启。问了相关人员、查了相关文档才明白。没开启ck的时候，状态...

2019-07-30 14:18:47 274

原创 Flink Keyby(Java-bean)类型

最近在改业务代码，以前都图省事了全部用tuple，代码经常出现DataStream<Tuple2<Tuple7<String,String,String,String,String,String,String>,Integer>>，发现后期很难维护，时间久了，也不一目了解每个string代表什么意思。全部改为Java POJO方式。例如下面代码 Dat...

2019-07-24 10:43:44 3431

原创 Flink-Sql自定义UDF

最近尝试使用flink的table-sql，发现没有from_unixtime函数，只能自定义该udf。原始kafka消息日志{"action":"exposure","itemId":"16c65063e51d4d834722bf1a4b1d6378@TT@1576","rankIndex":14,"time":"1563641998","unionId":"ohmdTtymqiQw5aSx...

2019-07-22 15:24:21 7405

原创 brew udpate更新

因为墙的原因，brew无法更新fatal: unable to access 'https://github.com/Homebrew/homebrew-cask/': Empty reply from serverfatal: unable to access 'https://github.com/Homebrew/brew/': LibreSSL SSL_connect: SSL_ERR...

2019-07-16 15:51:08 1325

原创 Debug小技巧

最近看<<码出高效>>这本书.里面提到几个细节，感觉很好。例如,日志级别设置WARN，但针对DEBUG级别日志，如果在程序中写出logger.debug("Processing trade with id:" + id + " and symbol:" + symbol);虽然日志不会被打印，但是会执行字符串拼接做出，如果symbol是对象，还会执行toString...

2019-06-26 14:50:44 189

原创 Flink入门(十六) State

Flink的state主要是有两种:Keyed State主要依赖于KeyedStreamOperator Statenon-keyed state这两种存在于两种形式：Managed State(由flink-runtime控制的结构化数据)Raw State。举例说明Managed State使用方法。//用户点击方案的事件操作DataStream<ItemI...

2019-06-24 15:05:49 360

原创 Spark常用函数(java)

//窗口row_number使用import org.apache.spark.sql.expressions.Window;import org.apache.spark.sql.expressions.WindowSpec;WindowSpec w = Window.partitionBy("start_station_code", "end_station_code").orderBy...

2019-06-20 15:55:57 721

原创 Flink入门(十五) 写入hdfs文件csv格式，自定义StreamWriterBase

上篇文章Flink入门(七) 写入HDFS现在，我想把java-bean写入csv文件。有两种方式，第一种将增加map，将bean改成string类型（以“，”分割），但不可以复用。第二种方式，重写StreamWriterBase，类似于StringWriter。代码如下import org.apache.flink.api.java.io.CsvOutputFormat;import...

2019-06-19 14:39:47 3355 3

原创 Flink入门(十四)大job的StateBackend压力测试

最近有这样需求，两个topic消息interval-join。其中一个topic是，展示列表的详细信息（曝光），大约20分钟，有100G大小（主要从服务端发送）；另外一个topic是，用户操作列表（点击、下单），由前端发送，数据很少，大约10分钟几十M，action表示操作，itemId表示方案的唯一标识符，itme表示操作时间，unionId表示用户唯一标识符。{"action":"cl...

2019-06-19 12:18:32 1610 4

原创 Flink入门(十三) 读RocketMQ-Flink消息

github已经有开源地址RocketMQ-Flink直接把代码复制到自己代码中这里面我新建一个类org.apache.rocketmq.flink.common.serialization.SimpleStringDeserializationSchemapackage org.apache.rocketmq.flink.common.serialization;import or...

2019-05-28 16:39:13 13810 18

原创 Flink入门(十二)自定义eventTime

kafka消息是自带timestamp的，但有的时候需要自定义eventTime，直接上代码 final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(); env.setParallelism(2); //这里我采用eventTi...

2019-05-27 15:17:04 2936 3

原创 Flink入门(十一) state状态update更新

有这样需求，统计每个小时，一个线路被点击次数，需要每30秒统计一次写入redis，采用Slide窗口滑动，但发现一个问题，当key没有新的消息消费，当（key，value）的value不变的，滑动统计还是会输出统计值，这个key的value就没必须要写入redis。所以采用ValueState的update更新，代码如下： DataStream<Tuple2<S...

2019-05-23 19:55:58 5866 2

原创 Flink入门(十)异步多线程Mysql打宽表，加维度

有这样需求，消息发来是有站点名，但与需要和mysql中维度表join出站名名的三字码（唯一标识符）。flink版本1.6.3，maven配置如下： <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</a...

2019-05-07 16:01:33 8629 1

原创 Flink入门(九)batch读csv文件

原数据结构parent_order_no,member_id,union_id,extra_cost,create_dateTW168D72A322BWWAA6KGP1608,100001608,ohmdTt8m7-j8VXCvHhG2LFVTlNrs,40.0,2019-03-08TW1694B5C000BPDWOJG5D2464,100002464,ohmdTtyEE9Rh4T-ZPq0...

2019-05-05 16:15:15 4079 3

原创 Pandas使用技巧

记录常用的pandas操作import pandas as pd#读csv文件data= pd.read_csv(r'D:\station_choose.csv',encoding='gbk')#读excel文件data= pd.read_excel(r'D:\前1000查询线路.xlsx',encoding='gbk')#查看列数和行数data.shape#查看类型d...

2019-04-23 15:10:05 277

原创 Spark解决task任务运行时间过长

今天补数据的时候，发现突然任务运行时间比以往过长。查看运行状态，发现有一个task运行过长。解决方法，开启spark.speculation，配置如下：spark.speculation truespark.speculation.interval 100：检测周期，单位毫秒；spark.speculation.quantile 0.75：完成task的百分比时启动推测；spark.s...

2019-04-18 14:21:39 10625

原创 Spark-ml模型保存为PMML

spark版本2.1.3maven设置 <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-mllib_2.11</artifactId> <version>2.1...

2019-04-03 17:01:58 3875 15

原创 Spark-ml交叉验证demo

2019-04-03 15:45:37 820

原创 Spark-Sql数组array类型转string

2019-04-02 15:07:38 11657 2

原创 Spark多行合并一行collect_list使用

有这样需求，原始数据如下图。开发需要把start_city_id和end_city_id作为key，value是一个list集合，里面包含这些所有所有字段详细信息，存到redis。 |-- first_traffic_type: string (nullable = true) |-- first_traffic_code: string (nullable = true) |-- fi...

2019-03-26 11:21:16 5207

原创 Flink入门(八) checkpoint与savepoint区别

checkpoint是自动的，一般30s触发一次，如果在控制台cancel任务，checkpoint都会被清空。默认checkpoint功能是disabled的。savepoint是手动的，适合场景大的窗口或则是状态任务，让每次启动要重复计算好多数据 flink savepoint b9a0bd873d0546e9d4511347c8031b74 /tmp/flink/savepoints/...

2019-03-22 10:35:10 1717

原创 ES拼音中文智能提示suggest

安装IK中文分词和拼音插件./bin/elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v6.3.2/elasticsearch-analysis-ik-6.3.2.zip./bin/elasticsearch-plugin install htt...

2019-03-19 14:59:47 5857 1

原创 Spark读CSV文件

spark可以自动csv文件判断字符类型原文件内容如下：parent_order_no,member_id,union_id,create_time,create_dateTW168E93658FBHBSATPTX9866,33459866,ohmdTt1gcNVNORpm_onak1nOTduE,2019-02-14 07:34:39,2019-03-07TW169383A50F2RPX...

2019-03-07 14:25:26 5157