- 博客(290)
- 资源 (1)
- 收藏
- 关注
RSS订阅原创 PysparkNote006---pycharm加载spark环境
外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cNRvCHDO-1690362477564)(https://github.com/weigetang406team/pic_set/blob/main/2023Q1/04_pycharm_with_pyspark.jpg?raw=true)]2023-07-26 阴 于南京市江宁区。pycharm配置pyspark环境，本地执行pyspark代码。spark安装、添加环境变量不提了。
2023-07-26 17:08:31
1039
原创 PysparkNote006---rdd的flatMap
在有些场景下，比如对df做操作，需要保留某些列，并且做些列之间复杂计算时，需要把最终结果再还原成dataframe时，比较好用。2023-07-26 阴 于南京市江宁区。flatmap()是将函数应用于RDD中的每个元素，将返回的迭代器的所有内容构成新的RDD。map()是将函数用于RDD中的每个元素，将返回值构成新的RDD。写的比较抽象，可以看下面第二个blog，有图讲的清晰些。对于map来说，dataframe的一行是元素，
2023-07-26 15:35:35
104
原创 PyPackage01---Pandas16_apply返回多个值时的索引问题
pandas使用apply函数或者groupby函数返回多个值时，会出现返回多个索引的问题，介绍下解决方法。
2022-11-27 15:17:46
504
1
原创 PythonNote041---exec变量作用域问题
fun_y1 = 1{fun_y1 = 1}fun_y1 = 1{locals().fun_y1 = 1(fun_y1 = 1,fun_y1 = 1None)}fun_y1 = 1此时不会污染全局变量。
2022-11-27 12:11:16
380
原创 PythonNote039---argparse处理布尔型参数
argparse无法识别布尔型参数，参数传进来时，默认为字符串。字符串强制转布尔型时为True。
2022-11-26 14:38:40
458
原创 PysparkNote103---window滑窗
window滑窗函数的使用，主要是应用在一些统计场景。举例：统计下面每个人每15分钟点击次数。
2022-08-03 17:25:19
558
原创 Hive10---explode拆分多行
Intro hive explode操作import pysparkfrom pyspark.sql import SparkSession# 创建SparkSession对象，调用.builder类# .appName("testapp")方法给应用程序一个名字；.getOrCreate()方法创建或着获取一个已经创建的SparkSessionspark = SparkSession.builder.appName("pysaprk").getOrCreate()import pys
2022-04-25 15:14:53
1571
原创 Hive08---插入数据
分区表插入一条数据至指定分区表INSERT INTO test_db.table1 PARTITION(ds='20210922') VALUES ('123', 'abc')Ref[1] https://dwgeek.com/hive-insert-into-partition-table-and-examples.html/
2022-04-25 15:12:59
1699
原创 PyPackage01---Pandas15_query in 列表及str正则匹配
Introquery in的用法对 dataframe的某个列，执行正则匹配，筛选相应行query inimport pandas as pddf = pd.DataFrame({"x": [1, 2, 3], "y": ['ab', 'bc', 'cd']})df.query("x in [1,2]") x y 0 1 ab
2022-04-25 11:04:30
2334
原创 PysparkNote005---批量写入Redis
Intro 批量写redis，主要考虑两个点：数据分批-spark的foreachPartition批数据的分批插入redis,redis pipeline提交Code直接看代码import redisimport jsonimport randomimport pandas as pdimport functoolskey_list = [f"test_{str(i).zfill(3)}" for i in range(100)]value_list = [json.dum
2022-04-24 19:42:06
1850
原创 PysparkNote004---foreachPartition的参数传递
Intro pyspark批量写入数据库时，需要分批写入，批量写入时，只要建立一个连接，这样可以显著的提高写入速度。分批写入，容易想到foreachPartition，但是pyspark不能像scala那样df.rdd.foreachPartition(x=>{...})只支持df.rdd.foreachPartition(you_function)看下源码： def foreachPartition(self, f): """ Applie
2022-04-24 18:14:17
1677
原创 PysparkNote102---DataFrame常用操作2
https://zhuanlan.zhihu.com/p/349016831 重复数据筛查 满足以下功能:筛选出重复的行。对某一个字段，筛选出重复的值对某几个字段筛选出重复的值1.1 重复行from pyspark.sql import SparkSession# 创建SparkSession对象，调用.builder类# .appName("testapp")方法给应用程序一个名字；.getOrCreate()方法创建或着获取一个已经创建的SparkSessionspark =
2022-04-24 17:55:31
1457
原创 PythonNote037---python中列表合并、嵌套打平
多个列表合并、嵌套列表合并的操作多个列表合并多个列表合并时常规操作，不提x = [0,1]y = [2,3][*x,*y][0, 1, 2, 3]x+y[0, 1, 2, 3]嵌套列表打平import operatorfrom functools import reducea = [[1,2,3], [4,5,6], [7,8,9]]reduce(operator.add, a)[1, 2, 3, 4, 5, 6, 7, 8, 9]或者reduce(lambda
2022-04-03 11:09:49
915
原创 PythonNote036---python中字典合并
多个字典合并的操作不同keyx = {"id": 1}y = {"name": "jack"}{**x,**y}{'id': 1, 'name': 'jack'}相同key两个字典合并相同key会出现覆盖的情况x = {"id": 1}y = {"name": "jack","id":2}{**x,**y}{'id': 2, 'name': 'jack'}如果value是list格式，可把同个key的value进行合并，或者做其他操作x = {"id": [1]}y
2022-04-02 15:50:58
719
原创 PythonNote035---python中星号的作用
可迭代对象的应用一个*、迭代对象的解包可以对字符串、元组、列表等可迭代对象进行解包fruit = ['apple','pear','grape','banana']*x,y=fruitprint(f"x={x},y={y}")x=['apple', 'pear', 'grape'],y=banana可以把list最后一个元素赋给y，其余的赋给x对于元组同理*x,y=(1,2,3,4,5)print(f"x={x},y={y}")x=[1, 2, 3, 4],y=5一个*、可
2022-03-08 09:58:07
108
原创 Hive09---字符串拼接，转json等
Intro 常用hive字符串拼接函数，转json等操作import pysparkfrom pyspark.sql import SparkSession# 创建SparkSession对象，调用.builder类# .appName("testapp")方法给应用程序一个名字；.getOrCreate()方法创建或着获取一个已经创建的SparkSessionspark = SparkSession.builder.appName("pysaprk").getOrCreate()im
2022-02-16 13:51:02
6338
原创 PyPackage01---Pandas14_对类别型变量做映射
Intro有时候，需要把字符串当做代码执行，看个例子getattrGet a named attribute from an object; getattr(x, ‘y’) is equivalent to x.yimport pandas as pddf=pd.DataFrame({"x":[1,2,3]})getattr(df,"x").count()3getattr(df,“x”)相当于df.x如果想执行df.x*2怎么弄？evaleval("df.x*2")0
2022-01-14 10:09:43
291
原创 PythonNote034---把字符串当做代码执行
Intro有时候，需要把字符串当做代码执行，看个例子getattrGet a named attribute from an object; getattr(x, ‘y’) is equivalent to x.yimport pandas as pddf=pd.DataFrame({"x":[1,2,3]})getattr(df,"x").count()3getattr(df,“x”)相当于df.x如果想执行df.x*2怎么弄？evaleval("df.x*2")0
2022-01-13 21:11:36
207
原创 PythonNote033---操作Redis
Intro关注点：目前只看hashset内容增改删除常规操作就可类似mysql，怎么减少长连接的操作增改删查import redisdb = redis.Redis(host='127.0.0.1', port=6379, password='12345', db=0, encoding='utf-8', decode_
2022-01-13 21:10:59
199
原创 FE01_OneHot-Python应用
Intro onehot的操作，解决两个问题：可以把列名关联上去可以处理之前的value数据构造import pandas as pddf=pd.DataFrame({"id":[1,2,3,4],"grade":["A","B","A","C"],"class":[ "1班","2班","3班","1班"]})pd.get_dummiescategory_feature=["grade","class"]pd.concat([df,pd.get_dummies(df[categ
2022-01-12 17:46:40
292
原创 Flink001---offset设置窗口起点
Intro滚动窗口，想要设置窗口开始的时点，怎么弄。举例说明：watermart设置为3s滚动窗口长度设置为5s起点设置为3，即[3,8)是一个窗口Code代码没啥说的，就是个offset的使用import org.apache.flink.api.common.functions.ReduceFunctionimport org.apache.flink.streaming.api.TimeCharacteristicimport org.apache.flink.streaming
2021-12-03 15:20:04
1481
原创 PyPackage01---Pandas13_比较两个df是否一致
Intro检查两个df是否完全一致import pandas as pdfrom pandas.testing import assert_frame_equalx1=pd.DataFrame({"x1":[1,2],"x2":[2,3]})x2=pd.DataFrame({"x1":[1,2],"x2":[2,3]})x3=pd.DataFrame({"x1":[1,2],"x2":[2,4]})def is_df_equal(df1,df2): try: ass
2021-11-18 15:50:06
1903
原创 Spark006---coalesce和repartition
Intro常用的重分区操作，简单记录下coalesce根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率。当 spark 程序中，存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减少分区的个数，减小任务调度成本默认情况下，不shuffle即增加分区数，没有变化减少分区数，会把该分区数据增加到其他分区中，原有分区数据保持不变import org.apache.spark.sql.SparkSessionimport org.apache.spark.rd
2021-11-18 15:15:09
1252
原创 Spark005---map、mapPartitions
Intromap、mapPartitions的使用和差异mapimport org.apache.spark.sql.SparkSessionval spark = SparkSession.builder().master("local[*]").getOrCreate()import org.apache.spark.sql.SparkSessionspark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSe
2021-11-17 14:17:23
1018
原创 Spark004-rdd分区逻辑
Infordd是怎么做分区切分的，即怎么把数据存放到各个分区中，直接看代码。Codeimport org.apache.spark.sql.SparkSessionval spark = SparkSession.builder().master("local[1]").getOrCreate()val dataRDD = spark.sparkContext.makeRDD(List(0，1, 2, 3, 4,5,), 4)dataRDD.foreachPartition(part =>
2021-11-16 14:50:22
765
原创 Spark003-基础概念
Infospark一些基本概念，主要参考尚硅谷-Spark教程从入门到精通，美团《Spark性能优化指南——基础篇》。DriverSpark 驱动器节点，用于执行 Spark 任务中的 main 方法，负责实际代码的执行工作。Driver 在 Spark 作业执行时主要负责：将用户程序转化为作业（job）在 Executor 之间调度任务(task)跟踪 Executor 的执行情况通过 UI 展示查询运行情况python代码在集群执行时，纯python部分主要在driver处执行，p
2021-11-16 14:49:04
733
原创 Mysql02---插表的各种操作
Info 批量写入mysql的两种情形：存在则删除，再全量写入存在则忽略，不存在则插入存在则更新，不存在则插入分别看下数据准备用python操作下，方便看操作之后结果。建表：CREATE TABLE `student` (`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',`stu_id` bigint(20) NOT NULL COMMENT '学号',`name` varchar(32) NOT N
2021-11-16 13:49:56
125
原创 Scala112-scala和mysql的交互
涉及到scala读、写、改mysql的一些操作，记录代码，备查。读读mysql一般有两种操作：整张表load进来通过执行sql，load部分数据进来import java.sql.{Connection, DriverManager, PreparedStatement, Timestamp}import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}def loadMysqlAsDf(sparkSession: Sp
2021-10-19 17:39:19
229
原创 Mysql01---一个bug引发的血案
不小心搞了个bug，差点造成严重事故，记录这血泪教训。背景CREATE TABLE `test` ( `id` BIGINT(20) NOT NULL AUTO_INCREMENT COMMENT '自增id', `name` string NOT NULL DEFAULT '' COMMENT '姓名', `age` int NOT NULL DEFAULT 0 COMMENT '年龄') 目的：修改name="小明"的age值，age=18过程错误写法UPDATE t
2021-09-26 20:44:58
94
原创 PythonNote032---python打印带颜色字符串
Intro 网上资料很多了，不多说，也有其他方法。记录下自己封装的小函数，方便自己的使用。Codedef magic_str(raw_str, display_mode="", font_color="", back_color=""): """ jupyter打印时可以显示颜色 :param raw_str: 原始字符串 :param display_mode:0默认1粗体4下划线5闪烁7反白显示 :param font_color:字体色-30黑31红32
2021-09-14 19:46:42
165
原创 PythonNote031---python和hdfs的交互
目的 有些简单的模型，希望直接python训练完模型，放在hdfs上，预测时，scala或者sql做特征，再用python进行预测，预测结果插入hive表。那么就涉及到如何加载hdfs的模型，遇到了些坑，只给出目前的解决方案，不求甚解~方法连接hdfs的python库有多个，用下来，hdfs安装相对方便，使用尚能满足持久化模型也有多种方法，pickle、joblib、pmml，joblib加载模型报错，暂时没找到解决方案，pickle可用from hdfs import InsecureCl
2021-08-25 16:08:30
219
原创 PyPackage01---Pandas12_print dataframe
print dataframe时的参数控制，index和列间隔。直接看代码~去除indeximport pandas as pdpd.__version__'1.1.5'df = pd.DataFrame({"id":[1,2,3],"name":["jordan","kobe","duncan"],"team":["Bulls","Lakers","Spurs"]})print(df) id name team0 1 jordan Bulls1
2021-06-08 16:56:41
139
winutils.exe&hadoop.dll
2019-12-07
空空如也
TA创建的收藏夹 TA关注的收藏夹
TA关注的人