一、需求背景
- 低级API：是和AWS的HTTP接口一一对应的，通过boto3.client(“xx”)暴露；
- 高级API：是面向对象的，通过boto3.resource(“xxx”)暴露，易于使用但是美中不足不一定覆盖所有API。
二、Pyspark
from pyspark.sql import SparkSession, SQLContext
from pyspark.sql import Window
ACCESS_KEY_ID = 'xxx'
SECRET_ACCESS_KEY = 'xxx'
END_POINT = 'xxx'
spark = SparkSession \
.builder \
.appName("write to S3") \
.enableHiveSupport() \
.config("spark.debug.maxToStringFields", "500") \
.getOrCreate()
hadoopConf = spark.sparkContext._jsc.hadoopConfiguration()
hadoopConf.set("fs.s3a.endpoint", END_POINT)
hadoopConf.set("fs.s3a.access.key", ACCESS_KEY_ID)
hadoopConf.set("fs.s3a.secret.key", SECRET_ACCESS_KEY)
sql = '''
hive sql取数
'''
bucket_name = 'xxx'
bucket_path = f"s3a://{bucket_name}"
spark.sql(sql).repartition(1).write.mode("overwrite").parquet(bucket_path)
跟上面类似，只需要把最后的写入改为读即可
hive_path = 'xxx'
df = spark.read.parquet(bucket_path)
df.write.parquet(f'{hive_path}', mode='overwrite')
比较麻烦的一点是要注意，写入s3时候的parquet文件只能控制文件夹的名字，而里面的文件名往往是类似’part-00000-c6089499-0ad2-4ff8-aae4-7c64f291c728-c000.snappy.parquet’这样的。这是由于文件多线程写入时，hive为了保证唯一性。暂时未找到方法解决。如果使用 toPandas().to_csv()中途可能会报内存不足GC
三、Boto3读写s3上的文件
这里主要是用的是boto3来处理，比较好用
import json
import boto3
with open('credentials.json', 'r') as fd:
credentials = json.loads(fd.read())
s3 = boto3.resource('s3',
endpoint_url=credentials['endpoint_url'],
aws_access_key_id=credentials['access_key'], aws_secret_access_key=credentials['secret_key'])
bucket = s3.Bucket(credentials['bucket_name'])
for obj in bucket.objects.filter(Prefix='xx'):
key = obj.key
bucket.download_file(Filename='xx.parquet',
Key=key)
bucket.upload_file(Filename='xx.parquet',
Key=key)
如果不想让文件下载到本地占用空间，可以直接object转为dataframe，速度也很快
import io
key = 'key_in_s3'
buffer = io.BytesIO()
s3.Object(bucket_name=credentials['bucket_name'], key=key).download_fileobj(buffer)
df = pd.read_parquet(buffer)
上传这里也可以直接dataframe上传，但是类似的api我一直传上去是空的，无奈用了简单API
up_buffer = io.BytesIO()
df.to_parquet(up_buffer, index=False)
s3.Object(bucket_name=credentials['bucket_name'],\
key='xx.parquet')\
.put(Body=up_buffer.getvalue())
也有一些其他的用法，这里大概列举一些
import boto3
s3 = boto3.resource("s3")
bucket = s3.create_bucket(Bucket="my-bucket")
for bucket in s3.buckets.all():
print(bucket.name)
s3.buckets.fitler()
bucket = s3.Bucket("my-bucket")
bucket.name
bucket.delete()
bucket.download_file(Key, Filename, ExtraArgs=None, Callback=None, Config=None)
bucket.objects.all()
objects = bucket.objects.filter(
Delimiter='string',
EncodingType='url',
Marker='string',
MaxKeys=123,
Prefix='string',
RequestPayer='requester',
ExpectedBucketOwner='string'
)
obj = bucket.Object("xxx")
obj = s3.Object("my-bucket", "key")
obj.delete()
obj.download_file(path)
with open('filename', 'wb') as data:
obj.download_fileobj(data)
rsp = obj.get()
body = rsp["Body"].read()
obj.upload_file(filename)
obj.upload_fileobj(fileobj)
其他的API使用可以查阅reference的官方文档，建议只阅读高级API部分即可
Reference
Original: https://blog.csdn.net/weixin_41733400/article/details/124654892
Author: 建微知筑
Title: Python利用boto3以及Pyspark操作AWS S3
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/818138/
转载文章受原作者版权保护。转载请注明原作者出处！