跳转至

MinIO

一、概述

数据湖（Data Lake）是一种基于集群和分布式文件系统的一种数据存储方式，对企业中的所有数据进行统一存储，数据湖的就是原始数据保存区，从原始数据（源系统数据的精确副本）转换为用于报告、可视化分析和机器学习等各种任务的目标数据。 数据湖中的数据包括结构化数据（关系数据库数据），半结构化数据（CSV、XML、JSON等），非结构化数据（电子邮件，文档，PDF）和二进制数据（图像、音频、视频），从而形成一个容纳所有形式数据的集中式数据存储。

数据湖的主要供应商一般都是超大规模公共云供应商，例如亚马逊AWS、微软Azure和谷歌云平台(GCP)。

AILab-PDBC (Python DataBase Connectivity)是数智教育发展(山东)有限公司 AI Lab 100 团队开发的高效、灵活的数据接口(API)

ailab100.pdbc.datalake类用于数据湖的读写,支持MinIO 、亚马逊Amazon S3、谷歌GCS、微软Azure Blob等主流数据湖。

本章介绍如何使用ailab100.pdbc.datalake.MinIO 连接 MinIO 进行读写和下载。

MinIO 是一个基于 Golang 实现的高可用、高弹性的开源对象存储系统（OSS），皆在提供云上的高可扩展与高性能的分布式文件存储系统。 MinIO 完全兼容 Amason 的 S3 分布式对象存储项目。

Amazon S3 是一种对象存储服务，采用分布式架构，数据存储在多个物理位置以提高可靠性和容错能力。S3使用容器化技术，可以实现水平扩展，从而满足不同规模和需求的存储。

Google Cloud Storage(GCS) 是 google 公司的一个存储平台，可提供高性能的对象存储服务，并包含出色的可伸缩性，数据可用性，持久性和安全性。它使您可以存储对象并立即从任何存储类别访问任何数量的数据，使用单个统一的API 将存储集成到应用程序中，并轻松优化价格和性能。

Azure Blob 存储是 Microsoft 提供的适用于云的对象存储解决方案。 Blob 存储最适合存储巨量的非结构化数据。

二、API 接口说明

alcedo_pdbc.datalake.MinIO

ailab100.alcedo_pdbc.datalake.MinIO 类用于创建 MinIO S3实例，实现MinIO 的读、写、上传、下载等操作。

Parameters:

Name Type Description Default
endpoint str

S3兼容对象存储服务endpoint。

 required
access_key str

对象存储的Access key。（如果是匿名访问则可以为空）。

 required
secret_key str

对象存储的Secret key。（如果是匿名访问则可以为空）

 required
secure bool

默认值为True 设为True代表启用HTTPS

 False
region str

可选，默认值为None 设置该值以覆盖自动发现存储桶region。

 None

Examples:

>>> from alcedo_pdbc.datalake import MinIO
>>> minio_client = MinIO('127.0.0.1:9000',access_key='Q3AM3UQ867SPQQA43P2F',
secret_key='zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG')
Notes

MinIO S3部署时会曝光两个端口9000和9001，其中9000为客户端连接端口，9001为Web管理端口。

Functions

alcedo_pdbc.datalake.MinIO.bucket_exists(bucket_name) method descriptor

MinIO.bucket_exists(self, str bucket_name: str) -> bool

bucket_exists()方法用于检查存储桶是否存在。

Parameters:

Name Type Description Default
bucket_name str

存储桶名称

 required

Returns:

Name Type Description
bool bool

如果存储桶存在返回True,否则返回False

Examples:

>>> if minio_client.bucket_exists("my-bucket"):
>>>     print("my-bucket exists")
>>> else:
>>>     print("my-bucket does not exist")
alcedo_pdbc.datalake.MinIO.download_file(minio_path=None, bucket=None, key=None, threads=4, local_path='.') method descriptor

MinIO.download_file(self, str minio_path: str = None, str bucket: str = None, str key: str = None, int threads: int = 4, str local_path: str = '.')

download_file()方法用于从 MinIO下载文件，可通过 s3_path 参数,或通过 bucket 和 key 参数

Parameters:

Name Type Description Default
minio_path str

MinIO 文件路径, 可选项, 与 bucket 和 key 参数不能同时为空.

 None
bucket str

MinIO 桶名, 可选项, 与 minio_path 参数不能同时为空,

 None
key str

带扩展名的文件名

 None
threads int

线程数

 4
local_path str

可选，文件保存路径，默认为当前目录 './'

 '.'

Returns:

Name Type Description
file ``CSV``、``Excel``、 ``JSON``、``HTML``、 ``HDF5``、 ``Feather``、 ``Parquet`` 、``Apache Avro``

根据参数filename文件类型导出文件保存至指定的目录下

Examples:

>>> minio_client.download_file(minio_path="s3://datalake/datasets/3财务困境研究数据集/ST财务预警.csv",threads=4)
>>> df = pd.read_csv('./ST财务预警.csv')
>>> print(df.info)
alcedo_pdbc.datalake.MinIO.download_folder(minio_path=None, bucket=None, key=None, local_path_to_download='.') method descriptor

MinIO.download_folder(self, str minio_path: str = None, str bucket: str = None, str key: str = None, str local_path_to_download: str = '.')

Takes s3 path or (bucket and key name) as arguments and download the folder

Parameters:

Name Type Description Default
minio_path str

S3 path from where it needs to download the folder. Defaults to None.

 None
bucket str

S3 bucket name, if S3 path is not provided . Defaults to None.

 None
key str

S3 Key name, if S3 path is not provied. Defaults to None.

 None
local_path_to_download str

save location. Defaults to '.' (current directory).

 '.'

Returns:

Name Type Description
file ``CSV``、``Excel``、 ``JSON``、``HTML``、 ``HDF5``、 ``Feather``、 ``Parquet`` 、``Apache Avro``

根据参数filename文件类型导出文件保存至指定的目录下
alcedo_pdbc.datalake.MinIO.get_bucket_policy(bucket_name, prefix) method descriptor

MinIO.get_bucket_policy(self, str bucket_name: str, str prefix: str)

bucket_exists()方法用于获取存储桶的策略

Parameters:

Name Type Description Default
bucket_name str

存储桶名称。

 required
prefix str

对象的名称前缀。

 required

Returns:

Name Type Description
Policy Policy

Policy枚举：Policy.READ_ONLY，Policy.WRITE_ONLY，Policy.READ_WRITE或 Policy.NONE。

Examples:

>>> policy = minio_client.get_bucket_policy(bucket_name="my-bucket",
                                           prefix='/datasets/1客户订单数据集')
alcedo_pdbc.datalake.MinIO.list_buckets() method descriptor

MinIO.list_buckets(self)

list_buckets()方法应用于返回所有可访问桶的信息

Returns:

Name Type Description
Buckets桶列表 list

class:Bucket <Bucket> object

Examples:

>>> buckets = minio_client.list_buckets()
>>> for bucket in buckets:
        print(bucket.name, bucket.creation_date)
alcedo_pdbc.datalake.MinIO.read_as_dataframe(minio_path=None, bucket=None, key=None, pandas_args={}, polars_args={}, extension='csv', return_type='pandas') method descriptor

MinIO.read_as_dataframe(self, str minio_path: str = None, str bucket: str = None, str key: str = None, dict pandas_args: Dict = {}, dict polars_args: Dict = {}, extension='csv', return_type='pandas')

read_as_dataframe()方法从 MinIO 读取 CSV 文件返回指定的 DataFrame。

注意：如果要加载多个文件或全部文件时，要求文件的数据结构相同。

Parameters:

Name Type Description Default
minio_path str

MinIO 文件路径，可选

与 bucket 和 key 参数不能同时为空. - 如果需要加载多个文件，请使用“bucket/path/filename*”； - 如果需要加载文件夹中的所有文件"bucket/folder/."

 None
bucket str

MinIO 桶名，可选

与 minio_path 参数不能同时为空,

 None
key str

带扩展名的文件名

 None
pandas_args dict

Pandas的参数

如encoding等

 {}
extension str

可选项，默认为 csv

文件的扩展名，自动从 minio_path 参数中获取。

 'csv'
return_type str

可选项，默认值为 pandas

返回 Dataframe 的类型，可选项为 pandas、polars 和 dask.

 'pandas'

Returns:

Name Type Description
DataFrame ``Pandas``、``Polars`` or ``Dask``

根据 return_type 参数返回对应的数据帧 Dataframe。

Examples:

>>> data=minio_client.read_as_dataframe(minio_path="/datalake/datasets/3财务困境研究数据集/ST财务预警.csv")
>>> print(data)
>>> #加载文件夹中的所有文件，要求数据结构相同
>>> data1=minio_client.read_as_dataframe(minio_path="/datalake/datasets/3财务困境研究数据集/")
alcedo_pdbc.datalake.MinIO.upload_file(source_file_path, bucket, key) method descriptor

MinIO.upload_file(self, str source_file_path: str, str bucket: str, str key: str)

Takes source file path, bucket and key as arguments and upload the file to S3

Parameters:

Name Type Description Default
source_file_path str

source file path

 required
bucket str

destination bucket

 required
key str

destination file path

 required
alcedo_pdbc.datalake.MinIO.upload_folder(local_folder_path, bucket, key) method descriptor

MinIO.upload_folder(self, str local_folder_path: str, str bucket: str, str key: str) -> None

Takes local path, bucket and key as arguments and upload the folder to s3

Parameters:

Name Type Description Default
local_folder_path str

local path of the folder want to be uploaded

 required
bucket str

s3 bucket name

 required
key str

s3 key name

 required
alcedo_pdbc.datalake.MinIO.write_dataframe(df, bucket, key, extension='csv', pandas_args={}, polars_args={}) method descriptor

MinIO.write_dataframe(self, df, str bucket: str, str key: str, extension='csv', pandas_args={}, polars_args={}) -> None

Takes DataFrame, bucket name, filename as arguments and write the dataframe to S3.

Parameters:

Name Type Description Default
df DataFrame

Dataframe which need to be uploaded

 required
bucket str

S3 Bucket Name

 required
key str

file name with extension

 required
extension str

extension of the files, It take automatically from the filename parameter.

 'csv'