数据读取、处理与保存

该部分为数据去读取、数据处理、数据保存以及对于文件的操作

数据读取（Extract）

Functions:

get_kml_df_with_level(gdf_dict)

从中gdf_dict读取数据以及文件夹名称,并构造文件夹层级,按照指定层级合并所 有数据,并打上层级标签

kml_df_create_level(gdf_dict)

读取 dict {level: gpd.GeoDataFrame} 字典, 并根据[name]列包含关系提取分辨文件夹层级

rdf(file_path, *[, sheet_name, ...])

常用文件读取函数，支持 .csv/.xlsx/.xls/.shp/.parquet/.pickle/.feather/.kml/.ovkml/ .geojson/.shp/.json等

rdf_by_dir(dir_path[, exts, ignore_index, ...])

从文件夹中读取所有文件并拼接成一个DataFrame

rdxls(file_path, *[, sheet_name, ...])

读取excel文件

read_all_sheets(file_path[, sheet_names, ...])

读取并合并所有sheet表

read_csv(file_path, *[, encoding, dtype, ...])

读取csv文件，基于 pd.read_csv，加入编码尝试

read_gdb(dir_path)

读取gdb文件夹

read_kml(file_path[, keep_level])

读取kml类文件,将其转换成DataFrame, 根据separate_folders选择是否拆分, 输出仅保留[name, geometry]字段,暂时只支持polygon和multipolygon数据

read_line_json(file_path[, encoding])

逐行读取json格式的文件

read_shapefile(file_path, **kwargs)

读取shapefile或geojson文件

read_shapefile_with_driver(file_path[, encoding])

读取shapefile文件，分别读取dbf中的属性表和shp中的geometry进行拼接

read_spss(file_path[, columns, nrows, encoding])

读取SPSS的.sav文件
ricco.etl.extract.get_kml_df_with_level(gdf_dict: dict) → GeoDataFrame

从中gdf_dict读取数据以及文件夹名称,并构造文件夹层级,按照指定层级合并所 有数据,并打上层级标签

参数:

gdf_dict – GeoDataFrame组成的字典，key为文件夹名称, value为对应数据：{level: gpd.GeoDataFrame}

ricco.etl.extract.kml_df_create_level(gdf_dict: dict) → dict

读取 dict {level: gpd.GeoDataFrame} 字典, 并根据[name]列包含关系提取分辨文件夹层级

参数:

gdf_dict – {dict} {level: gpd.GeoDataFrame} key为文件夹名称, value为对应数据

返回:

{‘板块名称’: 0, ‘T顶豪’: 1, ‘O1远郊品质’: 2}

返回类型:

文件夹层级字典，如

ricco.etl.extract.rdf(file_path: str, *, sheet_name=0, sheet_contains: str = None, encoding: str = None, info: bool = False, dtype=None, columns: (<class 'list'>, <class 'str'>) = None, nrows: int = None, recursive: bool = True) → DataFrame

常用文件读取函数，支持 .csv/.xlsx/.xls/.shp/.parquet/.pickle/.feather/.kml/.ovkml/ .geojson/.shp/.json等

参数:

  • file_path – 文件或文件夹路径

  • sheet_name – 数据所在sheet的名称，仅对.xlsx/.xls生效

  • sheet_contains – 筛选sheet名称中包含此字符串的sheet，仅对.xlsx/.xls生效

  • encoding – 编码

  • info – 是否打印数据集情况（shape & columns）

  • dtype – 指定读取列的类型

  • columns – 指定读取的列名

  • nrows – 指定读取的行数

  • recursive – 是否循环遍历更深层级的文件夹，默认为True，仅当路径为文件夹时生效

ricco.etl.extract.rdf_by_dir(dir_path, exts=None, ignore_index=True, recursive=False, columns: (<class 'list'>, <class 'str'>) = None, info=False) → DataFrame

从文件夹中读取所有文件并拼接成一个DataFrame

参数:

  • dir_path – 文件夹路径

  • exts – 要读取的文件扩展名

  • ignore_index – 是否忽略索引

  • recursive – 是否循环遍历更深层级的文件夹

  • columns – 指定列名

  • info – 是否打印基本信息（列名、行列数）

ricco.etl.extract.rdxls(file_path, *, sheet_name=0, sheet_contains: str = None, dtype=None, columns: (<class 'list'>, <class 'str'>) = None, nrows: int = None) → DataFrame

读取excel文件

参数:

  • file_path – 文件名

  • sheet_name – sheet表的名称

  • sheet_contains – sheet表包含的字符串

  • dtype – 指定读取列的类型

  • columns – 指定读取的列

  • nrows – 指定读取的行数

ricco.etl.extract.read_all_sheets(file_path, sheet_names=None, ignore_diff_columns=False)

读取并合并所有sheet表

参数:

  • file_path – Excel文件路径

  • sheet_names – 要读取并合并的sheet列表，默认读取全部的sheet

  • ignore_diff_columns – 如果每个sheet中的字段名不一致，是否要忽略

ricco.etl.extract.read_csv(file_path: str, *, encoding: str = None, dtype=None, columns: (<class 'list'>, <class 'str'>) = None, nrows: int = None, **kwargs)

读取csv文件，基于 pd.read_csv，加入编码尝试

ricco.etl.extract.read_gdb(dir_path)

读取gdb文件夹

ricco.etl.extract.read_kml(file_path, keep_level=True) → GeoDataFrame

读取kml类文件,将其转换成DataFrame, 根据separate_folders选择是否拆分, 输出仅保留[name, geometry]字段,暂时只支持polygon和multipolygon数据

参数:

  • file_path – kml or ovkml文件路径

  • keep_level – 是否保留文件夹层级标签

ricco.etl.extract.read_line_json(file_path, encoding='utf-8') → DataFrame

逐行读取json格式的文件

参数:

  • file_path – 文件名

  • encoding – 文件编码，默认为utf-8

ricco.etl.extract.read_shapefile(file_path, **kwargs)

读取shapefile或geojson文件

ricco.etl.extract.read_shapefile_with_driver(file_path, encoding='utf-8') → GeoDataFrame

读取shapefile文件，分别读取dbf中的属性表和shp中的geometry进行拼接

ricco.etl.extract.read_spss(file_path, columns=None, nrows=None, encoding=None)

读取SPSS的.sav文件

数据处理（Transform）

Functions:

convert_date(df, columns[, mode])

将日期转为当月的第一天或最后一天

convert_to_float(df, columns[, rex_method])

提取字符串中的数值信息并转为float类型

create_columns(df, columns[, value])

创建新的列，默认为空

df_iter(df, *[, chunksize, parts])

DataFrame切片生成器

dict2df(data[, c_key, c_value, as_index])

将字典转为dataframe

drop_by_df(df, df_deleted)

根据一个df_deleted中的数据从df中删除

expand_dict(df, c_src)

展开字典为多列

expand_dict_silent(df, c_src)

展开字典为多列

expand_graph(df, start_level[, c_key, ...])

将图数据展开为宽表

fillna_by_adding(df, col)

以自增的方式填充空值

filter_by_df(df, df_sizer)

从df中筛选出满足df_size的数据

fix_null(df[, columns])

将None/NaN/NaT等都替换为None

fuzz_df(df, col, target_series[, c_dst, ...])

模糊匹配。为DataFrame中的某一列从某个集合中模糊匹配匹配相似度最高的元素

is_changed(df_old, df_new[, key_cols, ...])

判断新旧数据集中的每一条数据是否变化

keep_best_unique(df, subset[, value_cols])

优化的去重函数，为保证数据的完整性，去重时优先去除指定列中的空值

null_if_in(df, columns, values)

如果列的值正好在values中定义的数组中，则将该列值设置为None

round_by_columns(df, columns)

对整列进行四舍五入，默认绝对值大于1的数值保留两位小数，小于1 的保留4位

split_list_to_row(df, column)

将列表列中列表的元素拆成多行

split_to_rows(df, column[, delimiter])

含有多值的列分拆成多行

table2dict(df[, key_col, value_col, orient])

DataFrame转字典

update_df(df, new_df[, on, overwrite, errors])

根据某一列更新dataframe里的数据

wrap_dict(df[, c_srcs, c_dst])

将指定的多列合并成一个字典列 :param df: 要处理的数据集 :param c_srcs: 需要包装成字典的多列, 如果不指定，则wrap所有的列 :param c_dst: 处理后的数据存储的列名称，默认为 'extra'
ricco.etl.transformer.convert_date(df: ~pandas.core.frame.DataFrame, columns: (<class 'str'>, <class 'list'>), mode: str = 'first') → DataFrame

将日期转为当月的第一天或最后一天

参数:

  • df – 要处理的DataFrame

  • columns – 要转换的列

  • mode – ‘first’ or ‘last’

ricco.etl.transformer.convert_to_float(df: ~pandas.core.frame.DataFrame, columns: (<class 'list'>, <class 'str'>), rex_method: str = 'mean') → DataFrame

提取字符串中的数值信息并转为float类型

参数:

  • df – 要转换的DataFrame

  • columns – 要转换的列

  • rex_method – 计算mean,max,min， 默认为mean

ricco.etl.transformer.create_columns(df: DataFrame, columns: list, value=None)

创建新的列，默认为空

参数:

  • df – DataFrame

  • columns – 列名

  • value – 值，默认为空

ricco.etl.transformer.df_iter(df: DataFrame, *, chunksize: int | None = None, parts: int | None = None)

DataFrame切片生成器

参数:

  • df – 要切片的DataFrame

  • chunksize – 每次返回的大小，与parts必须指定一个

  • parts – 返回的次数，与chunksize必须指定一个

ricco.etl.transformer.dict2df(data: dict, c_key='key', c_value='value', as_index=False)

将字典转为dataframe

ricco.etl.transformer.drop_by_df(df: DataFrame, df_deleted: DataFrame) → DataFrame

根据一个df_deleted中的数据从df中删除

ricco.etl.transformer.expand_dict(df: DataFrame, c_src: str)

展开字典为多列

ricco.etl.transformer.expand_dict_silent(df: DataFrame, c_src: str)

展开字典为多列

ricco.etl.transformer.expand_graph(df: DataFrame, start_level, c_key='id', c_level_type='level_type', c_parent_key='parent_id', c_info=None)

将图数据展开为宽表

参数:

  • df – 要展开的图数据

  • start_level – 开始展开的层级，一般是最低层级

  • c_key – 关键列

  • c_level_type – 层级类型列

  • c_parent_key – 父级关键列

  • c_info – 要保留的其他列

ricco.etl.transformer.fillna_by_adding(df: DataFrame, col: str)

以自增的方式填充空值

ricco.etl.transformer.filter_by_df(df: DataFrame, df_sizer: DataFrame) → DataFrame

从df中筛选出满足df_size的数据

ricco.etl.transformer.fix_null(df: ~pandas.core.frame.DataFrame, columns: (<class 'str'>, <class 'list'>) = None)

将None/NaN/NaT等都替换为None

ricco.etl.transformer.fuzz_df(df: ~pandas.core.frame.DataFrame, col: str, target_series: (<class 'list'>, <class 'pandas.core.series.Series'>), c_dst: str = None, valid_score=0) → DataFrame

模糊匹配。为DataFrame中的某一列从某个集合中模糊匹配匹配相似度最高的元素

参数:

  • df – 输入的dataframe

  • col – 要匹配的列

  • target_series – 从何处匹配， list/pd.Series

  • c_dst – 关联后输出的列名，默认为原列名+”_target”后缀

  • valid_score – 相似度大于该值的才返回

ricco.etl.transformer.is_changed(df_old: ~pandas.core.frame.DataFrame, df_new: ~pandas.core.frame.DataFrame, key_cols: (<class 'str'>, <class 'list'>) = None, value_cols: (<class 'str'>, <class 'list'>) = None, c_res: str = 'is_changed') → DataFrame

判断新旧数据集中的每一条数据是否变化

参数:

  • df_old – 原始数据集

  • df_new – 修改后的数据集

  • key_cols – 关键列，不指定则以索引列为准

  • value_cols – 要对比的列，默认出key_cols之外的其他列

  • c_res – 对比结果列名，默认为“is_changed”

ricco.etl.transformer.keep_best_unique(df: ~pandas.core.frame.DataFrame, subset: (<class 'list'>, <class 'str'>), value_cols: (<class 'str'>, <class 'list'>) = None) → DataFrame

优化的去重函数，为保证数据的完整性，去重时优先去除指定列中的空值

参数:

  • df – 要去重的Dataframe

  • subset – 按照哪些列去重

  • value_cols – 优先去除那些列的空值，该列表是有顺序的

ricco.etl.transformer.null_if_in(df, columns: (<class 'str'>, <class 'list'>), values: list)

如果列的值正好在values中定义的数组中，则将该列值设置为None

参数:

  • df –

  • columns – 需要检查的列，一个或多少

  • values – 所有需要标识为None的值

ricco.etl.transformer.round_by_columns(df, columns: list)

对整列进行四舍五入，默认绝对值大于1的数值保留两位小数，小于1 的保留4位

ricco.etl.transformer.split_list_to_row(df: DataFrame, column)

将列表列中列表的元素拆成多行

ricco.etl.transformer.split_to_rows(df: DataFrame, column: str, delimiter: str = '|')

含有多值的列分拆成多行

参数:

  • df –

  • column – 存在多值的列

  • delimiter – 多个值之间的分割符

ricco.etl.transformer.table2dict(df: ~pandas.core.frame.DataFrame, key_col: str = None, value_col: (<class 'str'>, <class 'list'>) = None, orient: str = 'dict') → dict

DataFrame转字典

参数:

  • df –

  • key_col – 生成key的列

  • value_col – 生成value的列

  • orient – 生成dict的方式，默认 ‘dict’,还有 ‘list’, ‘series’, ‘split’, ‘records’, ‘index’

ricco.etl.transformer.update_df(df: ~pandas.core.frame.DataFrame, new_df: ~pandas.core.frame.DataFrame, on: (<class 'str'>, <class 'list'>) = None, overwrite: bool = True, errors: str = 'ignore') → DataFrame

根据某一列更新dataframe里的数据

参数:

  • df – 待升级的数据集

  • new_df – 用于更新的DataFrame

  • on – （可选参数）用于判断更新哪些行的列

  • overwrite –

    （可选参数）控制如何处理原DataFrame在重叠位置上 非空 的值，默认为True

    • True: 默认值；使用 other DataFrame中的值覆盖原DataFrame中相应位置的值.

    • False: 只更新原DataFrame中重叠位置数据为 空 的值.

  • errors –

    （可选参数）控制如何处理两个DataFrame同一位置都有值的行为，默认为 ‘ignore’

    • ’ignore’: 默认值；DataFrame类型 df和other在同一个cell位置都是非NA值， 使用other中的值替换df中的值。

    • ’raise’: target和other都在同一位置包含非NA数据将抛出ValueError异常（’Data overlaps’）。

ricco.etl.transformer.wrap_dict(df: DataFrame, c_srcs: list | None = None, c_dst: str = 'extra')

将指定的多列合并成一个字典列 :param df: 要处理的数据集 :param c_srcs: 需要包装成字典的多列, 如果不指定，则wrap所有的列 :param c_dst: 处理后的数据存储的列名称，默认为 ‘extra’

数据保存（Load）

Functions:

to_csv_by_line(data, filename)

逐行写入csv文件

to_file(df, filepath, *[, index, log, encoding])

根据文件扩展名，将Dataframe保存为文件

to_parquet(df, filepath, index)

保存parquet文件，默认将shapely对象转换为wkb格式

to_parts_file(df, dirpath[, chunksize, ...])

将Dataframe保存为多个文件

to_sheets(data, filename[, index])

将多个dataframe保存到不同的sheet中
ricco.etl.load.to_csv_by_line(data: list, filename: str)

逐行写入csv文件

参数:

  • data – 要写入的数据列表

  • filename – 文件名

ricco.etl.load.to_file(df: DataFrame, filepath, *, index=False, log=True, encoding=None)

根据文件扩展名，将Dataframe保存为文件

参数:

  • df – 要保存的Dataframe

  • filepath – 文件路径，包含扩展名

  • index – 是否保存索引，默认不保存

  • log – 是否打印保存信息

  • encoding – 保存文件的编码

ricco.etl.load.to_parquet(df, filepath, index)

保存parquet文件，默认将shapely对象转换为wkb格式

ricco.etl.load.to_parts_file(df, dirpath, chunksize=None, parts=None, to_ext='.csv', **kwargs)

将Dataframe保存为多个文件

参数:

  • df – 要拆分保存的DataFrame

  • dirpath – 文件保存的目录

  • chunksize – 拆分保存的文件大小

  • parts – 拆分保存的文件数量

  • to_ext – 文件扩展名

ricco.etl.load.to_sheets(data: dict, filename: str, index=False)

将多个dataframe保存到不同的sheet中

参数:

  • data – 要保存的数据集，格式为：{sheet_name: DataFrame}

  • filename – 要保存的文件名

  • index – 是否保存index

文件操作（File）

Functions:

df_iter_by_column(df, by[, na])

按列的值分组迭代df

dir_file_to_x(from_dir, to_dir[, from_ext, ...])

转换整个目录中的文件格式

file_to_x(filepath, to_ext[, to_dir, ...])

文件格式转换，整体读取并转换，将文件整体加载到内存中再保存为另一种文件格式

gdb2csv(dir_path[, output_path, ...])

gdb文件转换为csv文件

merge_csv_files(dir_path, output_file)

合并指定目录下的所有CSV文件到一个输出文件中，极低内存消耗

reshape_files(from_dir, to_dir[, from_ext, ...])

将文件拆分成小文件，并保存到to_dir中

split2x(filename, *[, chunksize, parts, ...])

将文件拆分为多个文件，放置在与文件同名目录下

split2x_by_chunksize(filepath, *[, ...])

将文件拆分为多个文件，放置在与文件同名目录下

split2x_by_parts(filename, *[, parts, ...])

将文件拆分为多个文件，放置在与文件同名目录下

split_csv_by_column(filepath, by, *[, ...])

将csv文件按指定列的值拆分为多个文件
ricco.etl.file.df_iter_by_column(df: DataFrame, by, na='null')

按列的值分组迭代df

ricco.etl.file.dir_file_to_x(from_dir, to_dir, from_ext=None, to_ext=None, recursive=False, delete=False, log=False)

转换整个目录中的文件格式

ricco.etl.file.file_to_x(filepath, to_ext, to_dir=None, delete=False, overwrite=False, log=True, **kwargs)

文件格式转换，整体读取并转换，将文件整体加载到内存中再保存为另一种文件格式

参数:

  • filepath – 待转换的文件路径

  • to_ext – 要保存的文件扩展名

  • to_dir – 要写入的文件路径，默认存放在原目录

  • delete – 是否删除原文件，默认不删除

  • overwrite – 是否覆盖原文件，默认不覆盖

  • log – 是否输出文件保存、删除日志

ricco.etl.file.gdb2csv(dir_path, output_path=None, with_geometry=True, geom_format='wkb', log=True)

gdb文件转换为csv文件

ricco.etl.file.merge_csv_files(dir_path, output_file)

合并指定目录下的所有CSV文件到一个输出文件中，极低内存消耗

参数: - csv_dir: CSV文件所在的目录 - output_file: 合并后的输出CSV文件路径

ricco.etl.file.reshape_files(from_dir, to_dir, from_ext=None, to_ext=None, chunksize: int = 100000, func=None, log=False, **kwargs)

将文件拆分成小文件，并保存到to_dir中

参数:

  • from_dir – 读取的目录

  • to_dir – 保存的目录

  • from_ext – 读取的扩展名，默认为None，全部读取

  • to_ext – 保存的扩展名，默认为None，若读取文件的扩展名唯一，则使用读文文件的扩展名，反之需要指定

  • chunksize – 每个文件的大小

  • func – 自定义处理函数

  • log – 是否打印日志

ricco.etl.file.split2x(filename: str, *, chunksize: int | None = None, parts: int | None = None, to_ext='.csv', log=True, fn=None)

将文件拆分为多个文件，放置在与文件同名目录下

参数:

  • filename – 输入的文件路径

  • chunksize – 每份文件的数据量

  • parts – 输出的文件个数

  • to_ext – 输出的文件扩展名

  • log – 是否输出文件保存信息

  • fn – 处理函数

ricco.etl.file.split2x_by_chunksize(filepath: str, *, chunksize: int = 50000, to_ext='.csv', log=True, fn=None, **kwargs)

将文件拆分为多个文件，放置在与文件同名目录下

参数:

  • filepath – 输入的文件路径

  • chunksize – 每份文件的数据量

  • to_ext – 输出文件扩展名

  • log – 是否输出文件保存信息

  • fn – 处理函数

ricco.etl.file.split2x_by_parts(filename: str, *, parts: int | None = None, to_ext='.csv', log=True, fn=None)

将文件拆分为多个文件，放置在与文件同名目录下

参数:

  • filename – 文件路径

  • parts – 输出的文件个数

  • to_ext – 输出文件扩展名

  • log – 是否输出文件保存信息

  • fn – 处理函数

ricco.etl.file.split_csv_by_column(filepath: str, by: str, *, to_dir: str | None = None, to_ext: str = '.csv', merge_file: bool = True, chunksize=100000)

将csv文件按指定列的值拆分为多个文件

参数:

  • filepath – csv文件路径

  • by – 根据那一列拆分

  • to_dir – 要保存的位置，默认与原文件同名目录

  • to_ext – 输出的文件扩展名

  • merge_file – 是否将每一类数据合并为一个文件，默认合并

  • chunksize – 中间文件大小，如内存不足则可适当调低

数据标准化（熵值法）

Classes:

EntropyClass(df, cols, neg_cols, key)

Functions:

entropy(df[, columns])

采用熵值法计算的总分评价

gevalue(df, columns)

计算指标的差异系数

pca_score(df[, columns])

PCA法计算器

pvalue(df, columns)

计算指标下任一数值在该指标中的比重

rescale(df, key[, scale_min, scale_max, ...])

归一化处理

standard_e(df, columns)

熵值法的正向指标的极差标准化

standard_e_neg(df, columns)

熵值法的负向指标的极差标准化

wvalue(df, columns)

计算指标之间的权重关系
class ricco.etl.entropy.EntropyClass(df, cols, neg_cols, key)

基类：object

ricco.etl.entropy.entropy(df, columns: list | None = None)

采用熵值法计算的总分评价

参数:

  • df – 非负数化处理后的数据，默认其中数值量纲都是越大越好

  • columns – 需要用熵值法计算权重的指标合集

返回:

用过熵值法得到的总分, 权重

ricco.etl.entropy.gevalue(df: DataFrame, columns: list)

计算指标的差异系数

ricco.etl.entropy.pca_score(df, columns: list | None = None)

PCA法计算器

ricco.etl.entropy.pvalue(df: DataFrame, columns: list)

计算指标下任一数值在该指标中的比重

ricco.etl.entropy.rescale(df: DataFrame, key: str, scale_min: float = 0, scale_max: float = 1, score_range: tuple = (0, 100)) → DataFrame

归一化处理

参数:

  • df – 为原始待计算df

  • key – 需要参与计算的列名

  • scale_min – 下分位数, 低于该分位数的结果均取最小值

  • scale_max – 上分位数, 高于该分位数的结果均取最大值

  • score_range – 得分范围

ricco.etl.entropy.standard_e(df: DataFrame, columns: list)

熵值法的正向指标的极差标准化

ricco.etl.entropy.standard_e_neg(df: DataFrame, columns: list)

熵值法的负向指标的极差标准化

ricco.etl.entropy.wvalue(df, columns: list)

计算指标之间的权重关系

图数据处理（Graph）

Functions:

get_graph_dict(graph_df[, c_key, ...])

将图结构的Dataframe转为字典格式，提高查询效率

query_from_graph(key, graph_data[, c_key, ...])

从图数据中查询全部当前节点的全部父级节点
ricco.etl.graph.get_graph_dict(graph_df: DataFrame, c_key='id', c_level_type='level_type', c_parent_key='parent_id', c_parent_type='parent_type') → dict

将图结构的Dataframe转为字典格式，提高查询效率

参数:

  • graph_df – 图结构的数据

  • c_key – 索引列

  • c_level_type – 层级分类列

  • c_parent_key – 对应的父级索引列

  • c_parent_type – 对应的父级类型列

ricco.etl.graph.query_from_graph(key, graph_data: (<class 'pandas.core.frame.DataFrame'>, <class 'dict'>), c_key='id', c_level_type='level_type', c_parent_key='parent_id', max_depth=10) → dict

从图数据中查询全部当前节点的全部父级节点

参数:

  • key – 要查询的节点

  • graph_data – 要查询的数据集

  • c_key – 子节点关键列的字段名

  • c_level_type – 等级类型字段名

  • c_parent_key – 父节点字段名

  • max_depth – 最大深度