在 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)

一起來練習吧！

在 Python 中使用 Dask 進行平行程式設計