在 Dask bag 中使用任意数据

Python 中的 Dask 并行编程

James Fulton

Climate Informatics Researcher

复杂的混合数据格式

一个包含视频与音频的数据集示意图。

创建 Dask bag

import glob

video_filenames = glob.glob("*.mp4")

print(video_filenames)

['me_at_the_zoo.mp4', 'life_goes_on.mp4', 'guitar.mp4', 'hurt.mp4', ...]

创建 Dask bag

import glob

video_filenames = glob.glob("*.mp4")

import dask.bag as db

filename_bag = db.from_sequence(video_filenames)

filename_bag.take(1)[0]

'me_at_the_zoo.mp4'

加载自定义数据

# 加载单个视频
load_mp4("video.mp4")

{'video': array(
         [[[ 51,  57,  37, ..., 227, 238, 168],
         ...,
         [ 83, 125, 129, ..., 222, 148, 208]]]),
 'audio': array([   7. ,    9. ,    9.5, ..., -544.5, -551. , -558. ]),
 'filename': 'video.mp4'}

加载自定义数据

data_bag = filename_bag.map(load_mp4)

data_bag.take(1)[0]

{'video': array(
         [126, 162, 203, ...,  63,  58,   8],
         ...,
         [ 58, 222, 170, ..., 234,  63,  81]]]),
 'audio': array([-203.5, -209. , -207. , ..., -222.5, -233. , -248.5]),
 'filename': 'me_at_the_zoo.mp4'}

加载自定义数据

data_bag = filename_bag.map(load_mp4)

# 创建空列表
data_list = []

# 将延迟加载的文件添加到列表
for file in video_filenames:
    data_list.append(dask.delayed(load_mp4)(file))

延迟对象列表 vs. Dask bag

# 将延迟对象列表转换为 dask bag
data_bag = db.from_delayed(data_list)

# 将 dask bag 转换为延迟对象列表
data_list = data_bag.to_delayed()

进一步分析

transcribed_bag = data_bag.map(transcribe_audio)

transcribed_bag.take(1)[0]

{'video': array(
         [126, 162, 203, ...,  63,  58,   8],
         ...,
         [ 58, 222, 170, ..., 234,  63,  81]]]),
 'audio': array([-203.5, -209. , -207. , ..., -222.5, -233. , -248.5]),
 'filename': 'me_at_the_zoo.mp4'
 'transcript': "All right, so here we are in front of the, uh, elephants ...",
}

进一步分析

# 过滤无语音的视频
clean_bag = transcribed_bag.filter(transcript_is_not_blank)

# 对转录文本做情感分析
sentiment_bag = clean_bag.map(analyze_transcript_sentiment)

# 从 bag 中移除不需要的键
keys_to_drop = ['video', 'audio']
final_bag = sentiment_bag.map(filter_dictionary, keys_to_drop=keys_to_drop)

# 转为 Dask DataFrame
df = final_bag.to_dataframe()

结果

df.compute()

            filename              transcript        sentiment
0  me_at_the_zoo.mp4  All right, so here ...         positive
...              ...                     ...              ...

使用 .wav 文件

# 为 .wav 文件导入 scipy 模块
from scipy.io import wavfile

# 读取采样率与音频数组
sample_freq, audio = wavfile.read(filename)

使用 .wav 文件

# 每秒采样数
print(sample_freq)

# 音频数据
print(audio)

array([ 148,  142,  150, ..., -542, -546, -559], dtype=int16)

Vamos praticar!

Python 中的 Dask 并行编程