将数据帧发送到 Redis 的最快方法

Question

我有一个包含 2 列的数据框。对于每一行，我只想创建一个 Redis 集，其中数据框的第一个值是键，第二个值是 Redis 集的值。我已经进行了研究，我认为我找到了通过可迭代的最快方法：

def send_to_redis(df, r):
    df['bin_subscriber'] = df.apply(lambda row: uuid.UUID(row.subscriber).bytes, axis=1)
    df['bin_total_score'] = df.apply(lambda row: struct.pack('B', round(row.total_score)), axis=1)
    df = df[['bin_subscriber', 'bin_total_score']]
    with r.pipeline() as pipe:
        index = 0
        for subscriber, total_score in zip(df['bin_subscriber'], df['bin_total_score']):
            r.set(subscriber, total_score)
            if (index + 1) % 2000 == 0:
                pipe.execute()
            index += 1

有了这个，我每分钟可以向 Redis 发送大约 400-500k 集。我们最终可能会处理多达 3 亿个，按照这个速度，这需要半天左右的时间。可行但不理想。请注意，在外部包装器中，我一次从 s3 下载 .parquet 文件，并通过 IO 字节拉入 Pandas。

def process_file(s3_resource, r, bucket, key):
    buffer = io.BytesIO()
    s3_object = s3_resource.Object(bucket, key)
    s3_object.download_fileobj(buffer)
    send_to_redis(
        pandas.read_parquet(buffer, columns=['subscriber', 'total_score']), r)

def main():
    args = get_args()
    s3_resource = boto3.resource('s3')
    r = redis.Redis()
    file_prefix = get_prefix(args)
    s3_keys = [
        item.key for item in
        s3_resource.Bucket(args.bucket).objects.filter(Prefix=file_prefix)
        if item.key.endswith('.parquet')
    ]
    for key in s3_keys:
        process_file(s3_resource, r, args.bucket, key)

有没有办法在不使用迭代的情况下将此数据发送到 Redis？是否可以将整个数据块发送到 Redis 并让 Redis 为数据块的每个第一个和第二个值设置键和值？我想这会稍微快一些。

我在 Pandas 中使用的原始镶木地板是通过 Pyspark 创建的。我尝试使用速度非常快的 Spark-Redis 插件，但我不确定如何将我的数据转换为 Spark 数据帧本身内的上述二进制文件，而且我不喜欢如何将列名添加为字符串到每一个值，它似乎不是可配置的。每个具有该标签的 redis 对象似乎空间效率都很低。

任何建议将不胜感激！

Answer 1

试试Redis Mass Insertion 和redis bulk import using --pipe：

1. 创建一个包含 Redis 命令的新文本文件 input.txt

Set Key0 Value0
set Key1 Value1
...
SET Keyn Valuen

1. 使用redis-mass.py（见下文）插入redis

python redis-mass.py input.txt | redis-cli --pipe

---

redis-mass.py 来自 github。

#!/usr/bin/env python
"""
    redis-mass.py
    ~~~~~~~~~~~~~
    Prepares a newline-separated file of Redis commands for mass insertion.
    :copyright: (c) 2015 by Tim Simmons.
    :license: BSD, see LICENSE for more details.
"""
import sys

def proto(line):
    result = "*%s\r\n$%s\r\n%s\r\n" % (str(len(line)), str(len(line[0])), line[0])
    for arg in line[1:]:
        result += "$%s\r\n%s\r\n" % (str(len(arg)), arg)
    return result

if __name__ == "__main__":
    try:
        filename = sys.argv[1]
        f = open(filename, 'r')
    except IndexError:
        f = sys.stdin.readlines()

    for line in f:
        print(proto(line.rstrip().split(' ')),)