在大部分的项目中，使用spark streaming读取kafka的架构中，kafka的数据都是json格式发送的，比较复杂一些的是json中嵌套数组json格式，下面总结了两种解析方法 json数据格式 第一种方法 使用json4s的默认转化器解析 实现代码如下 第二种 使用spark.read.json解析数据 核心代码 依赖库配置 build.sbt
本文将详细介绍利用 ES 与 Hive 直接的数据交互；通过 Hive 外部表的方式，可以快速将hive数据导入到 elasticsearch 中，不需要编写代码只需要 Hive SQL 实现对数据的ETL 一、开发环境 1、组件版本 CDH 集群版本：6.3.2 ES 版本：7.7.0 Hive 版本：2.1.1 ES-Hadoop 版本：7.7.0 2、配置Hive 支持 ES-Hadoop 一共有四个方法可以配置，可以任选一个进行安装配置 （1）使用 add jar 在hive命令行或者hive s…
在工作或者学习中难免会遇到各种复杂需求的项目，有的是跟时间相关，有的跟距离相关等。 场景1 交通轨迹问题 数据有terminal_id,city,up_time三列。目标是统计car终端每次经过一个city的时间段；不是经过每个city的总时间。每个时间段以当地的up_time为准。 数据实例 需要实现获取的结果为 实现代码 ###pyspark实现代码 场景2 计算网站用访问用户的留存时间 例如有一个网站的用户登录数据如下 | user_name|login_date| +----------+---------…
在spark处理数据的项目开发过程中，经常会遇到清洗、转换数据的需求，转换数据就要在原来dataframe中增加新的字段，下面总结了3个常用增加字段的方法 首先创建一个dataframe数据实例 生成数据 方法1 使用withColumn()增加列 withColumn（）用于在DataFrame上添加新列或更新现有列，在这里仅说明如何使用现有列添加新列。 withColumn（）函数带有两个参数，第一个参数是新列的名称，第二个参数是Column类型中的列的值。 我们增加了一个新的字段 newsalary，字段值为…