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 ...",
}

さらなる分析

# 発話がない動画を除外
automatic_clean_bag = transcribed_bag.filter(transcript_is_not_blank)

# 文字起こしに感情分析を適用
sentiment_bag = automatic_clean_bag.map(analyze_transcript_sentiment)

# 不要な要素を削除
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ファイルの利用

# 1秒あたりのサンプル数
print(sample_freq)

# 音声データ本体
print(audio)

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

練習してみましょう！

Pythonで学ぶDaskによる並列プログラミング