- 博客(209)
- 资源 (4)
- 收藏
- 关注
RSS订阅
原创 整理一些HBase相关的文章
前言学习HBase过程中搜集的的一些文章，整理如下。文章列表HBase入门 – 数据模型HBase入门 – 组件HBase入门 – 读写与删除原理HBase入门 – 压缩、分裂与故障恢复HBase原理– 探索HFile索引机制HBase原理 – Memstore Flush深度解析HBase原理－数据读取流程解析HBase原理－迟到的‘数据读取流程’部分细节HBase...
2019-06-06 20:54:03
1047
原创 记录下本周末搭建个人博客的过程Mac+Hexo+GitHubPages
前言之前本来有一个个人博客，但是因为太懒没有维护，就来投奔CSDN了。这几天突然一时兴起，让好好弄一下自己的个人博客，因为CSDN的广告实在是…一言难尽…搜索了一般，选取一个比较简单的实现方式：即Hexo+GitHub Pages以下记录如果搭建个人博客网站 lestatzhang.com的过程具体步骤安装Git安装Node.js安装Hexo博客初始化将本地博客与GitHub关...
2019-05-26 17:15:44
2681
原创 为什么建议在Spark中使用Scala定义UDF
虽然在Pyspark中，驱动程序是一个python进程，但是它创建的SparkSession对象以及其他DataFrames或者RDDs等都是利用Python封装过的 JVM对象 。简单地说，虽然控制程序是Python，但它实际上是python代码告诉集群上的分布式Scala程序该做什么。 数据存储在JVM的内存中，并由Scala代码进行转换。将这些对象从JVM内存中取出并将它们转换为Pytho...
2019-03-19 15:36:37
5742
3
原创 关于Spark Dataset API中的Typed transformations和Untyped transformations
文章目录前言关于Dataset关于Dataset APITyped and Un-typed APIs关于TransformationsTyped transformations vs Untyped transformations总结参考资料前言学习Spark源代码的过程中遇到了Typed transformations和Untyped transformations两个概念，整理了以下相关...
2019-03-17 19:37:03
2651
原创 ETL工具Talend最佳实践
文章目录前言最佳实践前言和Talend这款软件打交道有一段时间了，主要用它来做一些ETL相关的作业开发，以下总结了一些自己开发配置与过程中的最佳实践。最佳实践可以通过修改Talend Studio 的 .ini 配置文件来给其分配更多的内存，例如，以下是我在64位8GB内存的电脑配置的参数-vmargs -Xms2014m -Xmx4096m -XX:MaxPermSize=...
2019-03-01 21:23:45
10586
4
原创 Impala中refresh与invalidate metadata
INVALIDATE METADATAINVALIDATE METADATA语句将一个或所有表的元数据标记为陈旧状态(stale)。下次Impala服务针对元数据无效的表执行查询时，Impala会在查询继续之前重新加载关联的元数据。由于与REFRESH语句完成的增量元数据更新相比，这是一个非常昂贵的操作，因此，在可能的情况下，建议使用REFRESH而不是INVALIDATE METADATA。在Impala之外，如在Hive和其他Hive客户端(例如SparkSQL)进行以下更改时，需要INVALIDA
2020-10-19 10:40:27
920
原创 spark动态资源分配参数一览
属性 默认值 含义 自版本 spark.dynamicAllocation.enabled false 是否使用动态资源分配，该资源分配将根据工作负载上下扩展在此应用程序中注册的执行程序的数量。 这需要设置spark.shuffle.service.enabled或spark.dynamicAllocation.shuffleTracking.enabled。 以下配置也相关：spark.dynamicAllocation.minExecutors，spa.
2020-07-14 16:03:07
2185
原创 Spark Dynamic Resource Allocation动态资源分配
动态资源分配Spark提供了一种机制，可以根据工作负载动态调整应用程序占用的资源。 这意味着，如果不再使用资源，应用程序可以将资源返还给群集，并在以后有需求时再次请求它们。 如果多个应用程序共享您的Spark集群中的资源，则此功能特别有用。默认情况下，此功能是禁用的，并且在各种类型的集群管理器中都可使用用。资源分配政策从较高的层次上讲，Spark应该在不再使用executor时将其放弃，并在需要它们时获取他们。由于没有确定的方法可以预测即将被删除的执行程序是否将在不久的将来运行任务，或者即将被添加的
2020-07-14 15:46:09
753
原创 shell同时后台运行多个进程并等待
代码如下job_list='job1 job2 job3'pids=''for j in `echo $job_list`;do run_job $j & pids="$pids $!"donefor pid in $pids;do wait $pid || let "RES=1"doneif [[ "$RES" -ne 0 ]]; then exit $RESfi
2020-06-15 12:08:05
2105
原创 Spark中对dataframe内重复列求和
前言在处理dataframe中的字段名是，发现有些字段名在处理后是重复，于是新定义了策略，对这些相同列名的字段进行求和合并summing the duplicated columns代码实现#Hanle Duplicatesfrom operator import addfrom functools import reduceimport collectionsoriginal_columns = [...]columns_after_cleanup=[...]counter=colle
2020-06-08 11:19:55
2148
原创 spark中递归join一系列dataframe
代码如下：def join_dfs(df_list,key=['id']): if len(df_list) == 1: retirm df_list[0] def join_df(df1,df2,key=['id']): return df1.join(df2, key) return reduce(join_df, df_list)def join_df_recursive(df_list, key=['id']): len_df = len(df_list) if len_df
2020-05-21 13:14:12
948
原创 注意pyspark中替换nan值的问题
在PySpark中，NaN与Null不同。 两者都与空字符串“”不同，因此，可能需要在所有数据集特定填充值的基础上检查每个字符。本来想使用df.replace(float('nan'),None)将dataframe中的 nan 值全部转化为空，但是测试下来发现，这个函数在spark中除了会将 nan替换掉，还会将 0 也替换成空（此处是int(0)，0.0不受影响），因此，要注意。最后，还是建议用isnan来判断然后替换：import pyspark.sql.functions as Fdf .
2020-05-21 09:42:18
2417
原创 pyspark中获取dataframe的大小
代码如下：from pyspark.serializers import PickleSerializer, AutoBatchedSerializerdef _to_java_obj_rdd(df): rdd=df.rdd._reserialise(AutoBatchedSerializer(PickleSerializer())) return rdd.ctx._jvm.org.apache.spark.mllib.api.python.Serde.pythonToJava(rdd._jrdd
2020-05-19 19:40:42
5660
原创 Hive中使用parquet
如何创建parquet外部表CREATE EXTERNAL TABLE IF NOT EXISTS XXXX (...)STORED AS PARQUETLOCATION '/data/hive/xxxx'TBLPROPERTIES ('parquet.compression'='SNAPPY')Parquet字段定义与Hive表中字段的关系如果hive表定义的字段包含且大于parquet文件字段，则parquet文件不存在的字段被查询为NULLparquet文件可以包含hive表定义
2020-05-19 14:27:16
1765
原创 python计算server时区时间差
记录一段python中关于timezone的小代码,将另一个时区服务器的时间转为utc+8import timefrom bson import ObjectIdfrom datetime import datetime, timedelta, timezonefrom dateutil.parser import parsedt1 = parse('2020-03-20')now_d...
2020-03-20 11:10:35
255
原创 hive中匹配反斜杠\
今天在做查询时候需要匹配某个字段中的反斜杠\，测试了一下，可以使用以下方法select name from my_table where name != regexp_replace(name,'\\\\','\\\\\\\\');本来想使用like 加 %来匹配的，但是一直无效，猜测是\转义的优先级比较高所以换一个思路，将每个字段中的’\‘的数量翻倍，这样，不含’\'的字段值不变。...
2019-12-09 21:42:50
3232
原创 python链表反转(递归)
代码如下：class ListNode(): def __init__(self,x): self.val =x self.next = None def printNode(head): curr=head while curr is not None: print(curr.val,end=' -&g...
2019-12-03 12:00:26
776
原创 python 实现链表反转操作
代码如下：class ListNode(): def __init__(self,x): self.val =x self.next = None def printNode(head): curr=head while curr is not None: print(curr.val,end=' -&g...
2019-11-30 17:51:58
244
原创 python实现堆排序
代码如下：def heap_adjust(arr, i, l): temp = arr[i] j = i*2 +1 while j< l: if j+1< l and arr[j]<arr[j+1]: j += 1 if arr[j]<temp: break ...
2019-11-30 17:08:46
179
原创 python实现树的前序/中序/后序遍历以及深度/广度遍历等
代码如下：class Node(): def __init__(self, x): self.val = x self.left = self.right = None class Tree(): def __init__(self): self.root = None def add(sel...
2019-11-30 16:48:56
276
原创 leetcode算法练习 —— 求两个排序排序数组的中位数
问题描述给定两个排序过的数组，求这两个排序数组的中位数， 在O(log(min(n, m)))内。解题思路将所有元素重组成有序的两部分，中位数就在段点处def find_median(a, b): #if len(a) > len(b): # a, b = b, a imin, imax = 0, len(a) n, m = len(a), len(b) while imi...
2019-10-15 21:34:34
260
原创 pyspark OneHotEncoder用法实例
def encode_columns(df, col_list): indexers = [ StringIndexer(inputCol=c, outputCol=f'{c}_indexed').setHandleInvalid("keep") for c in col_list ] encoder = OneHotEncoderEstimator( inputCols = [in...
2019-10-10 21:35:05
4781
原创 pyspark target mean encoding入门版
写了一个简单版本的target mean encoding， 代码如下：from pyspark.sql.functions import create_mapfrom itertolls import chainagg = df.select([f,target]).groupnBy(f).agg(avg_(target).alias('mean'), count_(target).ali...
2019-10-03 20:31:15
736
1
原创 Sparkexception: Values to assemble cannot be null
使用VectorAssembler的时候出现Sparkexception: Values to assemble cannot be null的错误，原因是因为Vector不能包含null值。解决方法如下：assembler.transform(df.na.fill(999999999))...
2019-09-30 15:06:52
853
1
原创 'requirement failed: Cannot have an empty string for name. '
今天测试pyspark重的encode出现了pyspark.sql.utils.IllegalArgumentException: 'requirement failed: Cannot have an empty string for name. '这个问题的主要原因是OneHotEncoder不能处理空字符串。所以我们需要将数据集中的空字符串提前处理一下。一般可以使用 dropna 或...
2019-09-30 13:50:57
1572
原创 python中的quicksort 和mergesort
def quicksort(alist): return quicksorthelper(alist, 0 , len(alist)-1) def quicksorthelper(alist, first, last): if first < last: splitpartition = partition(alist, first, last) ...
2019-09-24 11:57:55
435
原创 pyspark代码练习12 ——SparseVector & DenseVector
Vector对象可以转换为：Array [Double]使用toArray。使用toDense作为DenseVector的密集向量。使用toSparse作为SparseVector的稀疏向量。from pyspark.ml.linalg import Vectors#Create a dense vector (1.0, 0.0, 3.0).dv = Vectors.d...
2019-09-15 20:17:15
1600
原创 pyspark代码练习11 —— VectorAssembler
VectorAssembler是一个变换器，它将给定的列列表组合到一个向量列中。 将原始特征和由不同特征变换器生成的特征组合成单个特征向量非常有用，以便训练ML模型，如逻辑回归和决策树。 VectorAssembler接受以下输入列类型：所有数字类型，布尔类型和矢量类型。 在每一行中，输入列的值将按指定的顺序连接到一个向量中。from pyspark.ml.linalg import Vect...
2019-09-15 15:18:01
4416
原创 pyspark代码练习10 —— pipeline
from pyspark.ml.linalg import Vectorsfrom pyspark.ml import Pipelinefrom pyspark.ml.feature import OneHotEncoder, StringIndexerdf3 = spark.createDataFrame([ (Vectors.dense(10.2,), "a"), (Ve...
2019-09-14 20:32:17
726
原创 pyspark 代码练习9 —— StandardScaler
from pyspark.ml.feature import StandardScalerfrom pyspark.ml.linalg import Vectorsdf3 = spark.createDataFrame([ (Vectors.dense(10.2,), "a"), (Vectors.dense(1.6,), "b"), (Vectors.dense(23...
2019-09-14 20:29:17
1126
原创 pyspark 代码练习8 —— OneHotEncoder
不使用pipeline版本from pyspark.ml.feature import OneHotEncoder, StringIndexerdf = spark.createDataFrame([ (0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c")], ["id", "v...
2019-09-14 20:01:49
2831
原创 pyspark 代码练习7 —— 分层抽样
seed = 10sampleby = 'colA'fraction=0.8franctions = df.select(sampleby).distinct().withColumn('fraction', lit(fraction)).rdd.collectAsMap()print(fractions)sampled_df = df.stat.sampleBy(sampleby, f...
2019-09-11 12:26:41
1715
原创 pyspark 代码练习6 —— multiprocessing
import multiprocessing as mpdef f(x, y): return (x. y) pool = mp..pool(8)stats = pool.map(lambda x: f(x, y), x_list)pool.close()pool.join()
2019-09-09 21:46:58
552
原创 pyspark代码练习5 —— percentiles
import pandas as pddf_pd = pd.DataFrame( data = {'bools':[0,1,0], 'floats': [-1.0,0.5,2.7], 'doubles': [21.05, 458738453.34, 9089787.7] 'integers':[1,34,89]})df = spark.createDataFrame(df_pd...
2019-09-08 11:36:54
433
原创 pyspark代码练习4 —— 打印dtypes类型
from pyspark.sql import dtypesfor t in ['BinaryType', 'BooleanType', 'DateType', 'DecimalType','DoubleType','FloatType','IntegerType','LongType','ShortType','StringType','TimestampType']: print(f'{t...
2019-09-08 11:30:13
575
原创 pyspark 代码练习3 —— unique/skew/kurt
#n_uniqueagg_expr = [approxCountDistinct(col(c)).alias(c) for c in df.columns]unique_stats = df.agg(*agg_expr)unique_stats = unique_stats.withColumn('summary', lit('n_unique'))#skewagg_expr = [s...
2019-09-08 11:10:52
688
原创 pyspark 代码练习2
记录下最近写过的pyspark代码，如下：def sparkDFunion(df1, df2): cols1 = df1.columns cols2 = df2.columns total_cols = sorted( cols1 + list(set(cols2) - set(cols1))) def expr(mycols, allcols): ''' return th...
2019-09-07 23:03:57
563
原创 pyspark 代码练习1
记录下最近写过的pyspark代码，如下：edaStats = namedtuple('edaStats', 'is_numeric, n_unique, n_missing, _min, _25, _50, _75, _max, n_unique, n_freq, mean ')def eda_stats(dataframe, field_name, field_type): is_...
2019-09-07 22:55:22
522
原创 一个pipeline的例子
记录一个pipeline的例子，代码如下#Categrical features to pass down the categorical pipeline cateforical_features = ['date', 'waterfront', 'view', 'yr_renovated']#Numerical features to pass down the numerical p...
2019-08-25 22:36:48
718
原创 记录一个自定的CategoricalTransformer
前言自定义了一个简单的字符变量的变形器，代码如下。import numpy as np import pandas as pdfrom sklearn.base import BaseEstimator, TransformerMixinfrom sklearn.preprocessing import LabelEncoder, OneHotEncoder, StandardScale...
2019-08-25 21:56:31
250
空空如也
TA创建的收藏夹 TA关注的收藏夹
TA关注的人