当前位置：首页 > news >正文

专业网站开发工具网站广告条素材

news 2025/11/4 0:26:51

专业网站开发工具,网站广告条素材,wordpress获取分类下文章列表,帮忙推广的平台一、数据清洗，你都清洗什么？或者说 ETL 你是怎么做的？ 我在这个项目主要清洗的式日志数据，日志数据传过来的json格式去除掉无用的字段，过滤掉json格式不正确的脏数据过滤清洗掉日志中缺少关键字段的数据&#xff…

一、数据清洗，你都清洗什么？或者说 ETL 你是怎么做的？

我在这个项目主要清洗的式日志数据，日志数据传过来的json格式

去除掉无用的字段，过滤掉json格式不正确的脏数据

过滤清洗掉日志中缺少关键字段的数据（如缺少用户名，设备型号，等）

过滤清洗掉日志中不符合时间段的记录（由于app上日志上报有延迟等）

对用户的特殊信息进行脱敏，身份证，手机号等

将简单的的数据标准化（用户传过来的数据，经清洗后写入表格）

将重复的数据去重等（如session分割）

二、hive 和 spark 如何关联，关联过程中容易出现什么问题？

如何在DataGrip上使用hive的数据源编写Spark代码_datagrip配置spark-CSDN博客

如何关联：

启动hive的HiveMetaStore服务（hive-server-manager.sh start metastore）

设置spark的日志级别

启动spark的链接服务（10000端口hive在用，可以将spark的端口更改为10001）

遇到的问题：

hive中的数据库和sparksql中的数据库不是同一个，spark指向了本地的而不是集群的hdfs

链接之后，创建库表，上传数据后去集群的mysql中查看dbs中看这个库指向的是hdfs还是本地

需要修改hive的配置文件，然后复制到sparkconf然后分发集群重启就行

三、hive 如何处理 json 数据？

两个方法：

1、将数据视为无结构的string，将传入的json数据看作一条数据然后在hive建表上传

然后使用get_json_object函数

select get_json_object(line,'$.account') from test_json limit 10;

2、json格式进行映射（这需要jsonSerde的支持），创建表的时候加上ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.JsonSerDe'，然后就可读json数据，load就行

hdfs上的数据，导入表中一次后，直接删除，或者叫做移动，所以下一次再导入会说没有这个文件。

3、json_tuple() 可以一次提取多个字段。

select json_tuple('{"action_time":1682219447,"monitor_id":"0001","camera_id":"1","car":"豫A12345","speed":34.5,"road_id":"01","area_id":"20"}','camera_id','road_id','area_id','speed') as (cameraid,roadid,areaid,speed);

四、你的这个 spark 离线数仓的项目的项目架构图，请简单描述一下

数据经flum的抽取，将各个埋点的日志数据传入hive（进行处理sparksql处理）作为ods层，然后对ods层的数据进行清洗脱敏等然后当作tmp临时表，然后将tmp表进行合并生成dwd公共数据明细层，然后以dwd为基础，按天进行轻度汇总，粒度是一行信息代表的行为，例如一天下单的次数，以dws数据层为基础，按主题进行汇总，一个项目可以分为很多的主题，进行汇总，例如某个用户从注册开始到现在下单次数等，ads为各种统计表提供数据。

其中要对整个流程进行调度，监控，权限管理。

五、编写完成的 spark 程序，如何运行？

1、在本地模式运行

编写好的Python代码直接在本地运行

2、在集群环境中运行

编写完成的pyspark程序，上传到虚拟机，要把依赖的文件也上传上去，我这里上传的地址在/home/spark-shell中，然后将jsonsed的依赖jars包拉进去，然后sumit提交spark任务，传入参数，就能运行，在正常的情况下，要把每一py程序写成一个shell脚本，然后任务调度器每天定时调度shell脚本传入日期数据

spark-submit    \
--master yarn  \
--deploy-mode client \
--conf "spark.pyspark.driver.python=/opt/installs/anaconda3/bin/python3" \
--conf "spark.pyspark.python=/opt/installs/anaconda3/bin/python3" \
--driver-memory 512m \
--executor-memory 512m \
--executor-cores 1 \
--num-executors 2 \
--queue default  \
/home/spark-shell/AppLogWash.py $dt