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)

Ayo berlatih!

Python에서 Dask로 병렬 프로그래밍