sunlang的博客

原创 日志关键信息处理

从日志文件中根据关键词"dh=",筛选出符合要求的行。从行中筛选出列，这时一个单独的列就是一个字符串。从多行中筛选出最近一次的dh，也就是最后一样。spark任务异常失败后，需要重启任务。如何知道上次失败的时候dh执行到了何处。因此每次重启的任务这个参数都是不同的。用字符串处理方式得到最终的值。需要借助任务执行日志来确定。任务需要一个时间参数dh。dh是动态的向后流动的。

原创 azkaban二次开发

springboot封装azkaban的api，提供可调用azkaban任务流的接口流程如下：springboot接口->azkaban api->azkaban project(flow tasks)->shell脚本->spark tasks。

原创 Hive小文件处理

null

原创 任务流之间的调度依赖

无

原创 自定义函数

spark 中的 UDF (UserDefinedFunction) 大家都不会陌生, UDF 其实就是将一个普通的函数, 包装为可以按 “行“ 操作的函数, 用来处理 DataFrame 中指定的 Columns.例如, 对某一列的所有元素进行 +1 操作, 它对应 mapreduce 操作中的 map 操作. 这种操作有的主要特点是: 行与行之间的操作是 独立 的, 可以非常方便的 并行计算 每一行的操作完成后, map 的任务就完成了, 直接将结果返回就行, 它是一种”无状态的“但是 UDAF

原创 拉链表-spark版本

拉链表 spark

原创 长周期数据更新的处理

长周期数据更新的两种实现方案

原创 离线任务的稳定性

日志追踪，关键字提取，任务失败重启策略。

原创 基础同步工具

【代码】基础同步工具。

原创 基础工具类

【代码】基础工具类。

原创 开发环境之Spring.profiles.active

【代码】开发环境之Spring.profiles.active。

原创 变化的日期

【代码】变化的日期。

原创 flink+drools动态规则示例之温度跳变告警

flink+drools动态规则示例之温度跳变告警package com.mz.test;import com.google.common.collect.Lists;import lombok.*;import lombok.Getter;import lombok.NoArgsConstructor;import lombok.Setter;import lombok.ToString;import org.apache.flink.api.common.state.Broadcast

原创 docker部署maven私服

11-08-2020 11:38:53 CST t_uac_user ERROR - Job run failed!java.lang.RuntimeException: azkaban.jobExecutor.utils.process.ProcessFailureException: Process exited with code 1 at azkaban.jobExecutor.ProcessJob.run(ProcessJob.java:305) at azkaban.execapp.Jo.

原创 端口多次重试失败异常

11-08-2020 11:38:48 CST t_uac_user INFO - Starting job t_uac_user at 159711712868511-08-2020 11:38:48 CST t_uac_user INFO - job JVM args: '-Dazkaban.flowid=order' '-Dazkaban.execid=130' '-Dazkaban.jobid=t_uac_user'11-08-2020 11:38:48 CST t_uac_user INFO.

原创 flink状态与kafka消费参数

当没有使用状态的时候，如第一次部署我们设置了 consumer.setStartFromTimestamp(DateUtil.getMillsFromString(startTime));那么这个参数就会起作用当我们任务从ck状态恢复的时候，这个参数就失去了作用如下图任务从ck恢复，时间设置为2021-06-05但是并没有从凌晨消费，而是从下午3点...

原创 作为基础任务的窗口差值

package com.mz.iot.test;import com.mz.iot.utils.DateUtil;import com.mz.iot.utils.FlinkUtils;import com.mz.iot.utils.LogUtils;import lombok.*;import org.apache.commons.compress.utils.Lists;import org.apache.flink.api.common.functions.MapFunction;im.

原创 状态编程求窗口差值

package com.mz.iot.test;import com.mz.iot.optional.MyContinuousEventTimeTrigger;import com.mz.iot.utils.DateUtil;import com.mz.iot.utils.FlinkUtils;import com.mz.iot.utils.LogUtils;import lombok.*;import org.apache.commons.compress.utils.Lists;imp.

原创 能耗分析使用设备绝对值

package com.mz.iot.test;import com.mz.iot.optional.MyContinuousEventTimeTrigger;import com.mz.iot.utils.DateUtil;import com.mz.iot.utils.FlinkUtils;import com.mz.iot.utils.LogUtils;import lombok.*;import org.apache.commons.compress.utils.Lists;imp.

原创 云主机buff/cache清理脚本

#! /bin/bashfor i in hadoop-01 hadoop-02 hadoop-03do echo -------------- $i --------------- ssh $i " #!/bin/bash #每天0点清除一次缓存 echo "开始清理缓存" sync; #写入硬盘，防止数据丢失 sleep 10 #延迟10秒 echo 3 > /proc/sys/vm/drop_caches echo "清理结束""done.

原创 protobuf测试案例

最近物联网实时项目要进行分层以及加大供应商的数据传输频率，必然涉及到大量的网络io，宽带费用因此调研kafka数据传输和存储的压缩方案，决定采用protobuf环境依赖 <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId>

原创 docker安装kibana

各个环境都会用到的基础组件，使用docker来管理拉取kibana镜像docker pull kibana:6.5.4启动容器 docker run -d -p 5602:5601 kibana:6.5.4此时启动的容器，必然会报错，因为相关参数还没修改停止容器并进入终端修改参数docker stop 2622a72b6a97docker exec -it bold_williamson bash使用容器化这种方式的参数修改不像使用压缩包，只有几个必要的参数参

原创 Centos配置自己的服务开机自启

现在有很多套环境，每套环境都有需要用到的服务组件，比如kibana但是这个环境一旦重启，就得去手动启动这些服务为避免此麻烦，我们配置服务自启动，以kibana为例

原创 apache mr historyserver未生效问题排查

启动mr的历史服务后，虽然查看进程存在，但是yarn上的结束的任务总是看不到历史log查看历史服务启动日志，发现如下是因为hdfs权限的问题解决方案：要么手动创建好/tmp/logs目录，要么放开hdfs的权限...

原创 flink任务失败导致yarn宕机的排查

一，定位内存使用情况跟踪02节点上nodemanager的日志这个其实就是flink任务的jobManager的内存情况另外一个01上的日志这个就是flink任务的taskManager的内存情况暂时没有问题，任务正常，yarn正常这是因为本次提交增大了jobManager的内存，之前提交的时候只有1g内存，导致出现任务失败，yarn宕机...

原创 一些开发规范记录

数据采集规范由于需要采集的异构数据源有SQLSERVER，MYSQL，Elastic Search，单独的Sqoop不满足需求，因此选型Spark以自定义的方式采集数据到数据仓库导入策略为新增和变化ES的导入示例#!/bin/bashif [ $1 ];thenhost=$1elseecho 'need es host'exitfiif [ $2 ];thendt=$2elsedt=`date -d "-1 day" +%F`fiif [ $3 ];the

原创 es清空索引数据

POST ads_iot_latest_index/_delete_by_query{ "query": { "match_all": { } }}一个异常的解决Caused by: ElasticsearchException[Elasticsearch exception [type=cluster_block_exception, reason=blocked by: [FORBIDDEN/12/index read-only / allow delete.

原创 flink消费kafka偏移量的澄清

/** * 这个offset会保存在状态，也会保存在__consumer_offset中 * 因此如果没有使用新的消费者组，重启任务即使不使用状态也不会从头消费 */ consumer.setCommitOffsetsOnCheckpoints(true); /** * 这里始终不提交，在kafka-eagle中都不会有这个consumer的信息 */ ...

原创 flink的状态恢复之savepoint

savepoint使用场景状态恢复，程序更新，升级，bug修复，版本迁移，暂停或重启应用SavePoint是一个重量级的Checkpoint,你可以把它当做在某个时间点程序状态全局镜像，以后程序在进行升级，或者修改并发度等情况，还能从保存的状态位继续启动恢复。可以保存数据源offset，operator操作状态等信息，可以从应用在过去任意做了savepoint的时刻开始继续消费。用户手动执行，是指向Checkpoint的指针，不会过期，在集群升级/代码迁移等情况下使用。注意：为了能够在作业的

原创 kafka-eagle2.0.4编译以及填坑

编译，下载源码git clone https://github.com/smartloli/kafka-eagle.git用ide打开源码项目，配置好mavenproject的pom文件添加maven编译插件 <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId&

原创 maven仓库清理脚本

cls @ECHO OFF SET CLEAR_PATH=C: SET CLEAR_DIR=C:\Users\Administrator\.m2\repositorycolor 0a TITLE ClearLastUpdated For Windows GOTO MENU :MENU CLSECHO. ECHO. * * * * ClearLastUpdated For Windows * * * * ECHO. * * ECHO. * 1 清理*.lastUpdated * .

原创 kafka监控-eagle

版本下载http://www.kafka-eagle.org/articles/docs/changelog/changelog.html修改kafka-server-start.sh修改kafka-server-start.sh命令中if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"fi为if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then

原创 es生产索引升级实践-索引重建以及索引别名

索引升级方案一，索引重建：PUT /ads_iot_water_index_temp{ "mappings": { "_doc": { "dynamic": "strict", "properties" : { "cnts" : { "type" : "float" }, "dh" : { "type" : "text",

原创 es查询dsl使用sql自动生成

原创 峰谷均计算

package com.mz.iot;import com.alibaba.fastjson.JSON;import com.mz.iot.bean.MaxMinAvg;import com.mz.iot.bean.DataOrigin;import com.mz.iot.utils.Conf;import com.mz.iot.utils.DateUtil;import com.mz.iot.utils.EsSinkUtil;import com.mz.iot.utils.JsonUti.

原创 使用量统计

package com.mz.iot;import com.alibaba.fastjson.JSON;import com.mz.iot.bean.DataOrigin;import com.mz.iot.bean.UseResult;import com.mz.iot.utils.Conf;import com.mz.iot.utils.DateUtil;import com.mz.iot.utils.EsSinkUtil;import com.mz.iot.utils.JsonUti.

原创 flink的历史checkpoint点不可用的测试

程序升级，代码修改，重构，bug修复等等，当你的代码修改后，什么情况下会导致历史ck不可用?一，job graph变化会导致不可用例如注释掉流的print，print实际上是sink

原创 flink的窗口状态的计算

结论：flink的窗口计算是有状态的计算，并且是自动维护状态，持久化到外部系统以下是验证过程import com.mz.iot.utils.DateUtil;import org.apache.flink.api.common.functions.AggregateFunction;import org.apache.flink.api.common.functions.FilterFunction;import org.apache.flink.api.common.functions.

原创 flink窗口偏移

开滚动窗口的时候 /** * Creates a new {@code TumblingEventTimeWindows} {@link WindowAssigner} that assigns * elements to time windows based on the element timestamp and offset. * * <p>For example, if you want window a stream by hour,but window begi

原创 ssh_exchange_identification: Connection closed by remote host

待处理