Pythonで大量データ処理！ PySparkを用いたデータ処理と分析のきほん

PyConJP2017の資料　Python Spark PySpark PyConJP 2017 Apache Spark

1. 1ZUIPOͰେྔσʔλॲཧʂ
   1Z4QBSLΛ༻͍ͨσʔλॲཧͱ෼ੳͷ͖΄Μ
   
   1Z$PO+1
   
   $IJF
   )BZBTIJEB
   
2. ࣗݾ঺հ
   $IJF
   )BZBTIJEB
   
   5XJUUFS
   !DIJF
   
   3FUUZ*OD
   4PGUXBSF
   &OHJOFFS
   
   
   
   1ZUIPO
   ػցֶश
   )BEPPQ
   4QBSL
   4DBMB
   %#
   ج൫
   WJN
   ম೑
   ϐΞϊ
   ςχε
   εϊϘ
   
   
3. ࠓ೔͓͸ͳ͢͠Δ͜ͱ
   •  "QBDIF
   4QBSLͷ঺հ
   •  1Z4QBSLͷΞʔΩςΫνϟ
   •  4QBSLͷػցֶशϥΠϒϥϦʹ͍ͭͯ
   •  3FUUZʹ͓͚Δ4QBSLࣄྫʹ͍ͭͯ
   
4. ಥવͰ͕͢ʂ
   Pythonでデータ分析 してる人！
5. 1Z%BUBͷϥΠϒϥϦ܈
   ͱ͍͑͹ɺ
   ͱ͍͏͘Β͍ɺ1ZUIPO͸σʔλॲཧ΍෼ੳͷͨΊͷ
   ϥΠϒϥϦ͕ͦΖͬͯ·͢Ͷʂ
   ଞʹ΋ͨ͘͞Μʂ
   ศརJ
   
6. •  ୆ͷαʔόͰॲཧ͖͠Εͳ͍େن໛σʔλΛѻ͍͍ͨ
   •  σʔλྔ͕εέʔϧͯ͠΋ಈ͘࢓૊Έ͕΄͍͠
   •  େن໛σʔλʹର͢ΔػցֶशΛߦ͍͍ͨ
   
   ͱ͍ͬͨ৔߹ʹ͸ɺ1Z%BUBϥΠϒϥϦ܈Ͱ͸ରԠ
   ͖͠Εͳ͍৔߹͕͋Δɻ
   
   ͦΜͳͱ͖ʹ࢖͑Δͷ͕ɺ
   Ͱ΋ɾɾɾ
   
7. •  044ͷฒྻ෼ࢄॲཧϑϨʔϜϫʔΫ
   –  ॲཧதͷো֐࣌ͷϦΧόϦ΍ɺλεΫ෼ׂɾεέδϡʔϦϯάΛ
   
   
   
   
   
   4QBSL͕͏·͘΍ͬͯ͘ΕΔ
   –  αʔόͷεέʔϧΞ΢τʹΑͬͯεϧʔϓοτ͕ઢܗʹ͍͔ۙͨͪͰ޲্͢Δ
   •  Ϧιʔεར༻࠷దԽͷ޻෉͕͞Ε͓ͯΓɺॲཧ͕ߴ଎
   –  ΦϯϝϞϦϕʔεͷॲཧ
   –  +7.ͷΦʔόϔουΛվળ͢Δ1SPKFDU
   5VOHTUFO
   –  Ωϟογϡ
   –  ஗ԆධՁ
   
   •  1ZUIPOΛؚΉෳ਺ͷݴޠ͔Β࢖͑Δ
   –  ࠷৽ͷ4QBSLͩͱɺରԠόʔδϣϯ͸1ZUIPO  ʹରԠ
   •  ػցֶश΍ετϦʔϜॲཧɺॲཧͷྲྀΕ͕ݟ͑Δ6*ͳͲɺ
   
   
   
   
   
   ๛෋ͳػೳ͕͋Δ
   •  ίϛϡχςΟͷ׆ಈ͕׆ൃ
   "QBDIF
   4QBSLͱ͸ʂ
   
8. 4QBSLͷ๛෋ͳػೳ
   4QBSL
   $PSF
   4QBSL
   42-
   4QBSL
   4USFBNJOH
   ʢετϦʔϜॲཧʣ
   .MMJC
   ػցֶश
   (SBQI9
   άϥϑॲཧ
   SQL CSV {json} S3 BigQuery parquet Data Sources ༷ʑͳσʔλ ιʔεʹରԠ
   ෳ਺ݴޠΛαϙʔτ
   "1*͕ॆ࣮
   
9. Ϧονͳ6*
   DAG Visualiza?on: 処理の流れが見える Event Timeline: 各タスクの所要時間が見える Summary Metrics: 処理時間やデータ量などの メトリクス情報が見える •  λεΫͷਐߦঢ়گ΍ϝτϦΫε͕ݟ͑Δ6*͕͋Γɺσόοά͕΍Γ΍͍͢
   
10. ͍΍ɺͰ΋෼ࢄॲཧͬͯ
    ΊΜͲͦ͘͞͏ɻ
    ࠷ॳͷҰา͕౿Έग़ͤͳ͍ɻ
    ͬͯࢥ͏ਓ΋ଟ͍ͱࢥ͍·͢ɻ
    大丈夫、Sparkはサーバ1台でも動きます！
11. ࣮ࡍʹ΍ͬͯΈΑ͏ʂ
    $ wget hIps://d3kbcqa49mib13.cloudfront.net/spark-2.2.0-bin-hadoop2.7.tgz $ tar xzvf spark-2.2.0-bin-hadoop2.7.tgz $ cd spark-2.2.0-bin-hadoop2.7 $ bin/pyspark 4QBSL͔ΒɺQJQͰ΋ΠϯετʔϧͰ͖ΔΑ͏ʹͳΓ·ͨ͠ʂ
    ·ͨɺ%PDLFSΛ࢖͏ͷ΋ศརͰ͢ʂ
    μ΢ϯϩʔυɾΠϯετʔϧͯ͠
    ΠϯλϥΫςΟϒγΣϧΛ্ཱͪ͛Δ·Ͱ
    ͨͬͨίϚϯυʂ
    
12. •  )BEPPQΫϥελ্Ͱಈ࡞͢Δ
    •  "NB[PO
    &.3΍(PPHMF
    %BUB1SPDͳͲͷ
    ϚωʔδυαʔϏεΛ࢖͏ͱָ
    ຊ֨తʹ෼ࢄॲཧΛߦ͏ͱ͖ͷಈ࡞؀ڥ
    HDFS YARN MapReduce MesosやSpark Standaloneもある S3など他に様々な データソースにも対応する 並列分散処理 フレームワーク リソース管理、 ジョブスケジュール 分散ファイル システム Basic Stack SparkはMapReduceの 後継と言われる
13. 4QBSLͷͭͷϓϩάϥϛϯάϞσϧ
    3%%
    •  ίϨΫγϣϯૢ࡞ͷΑ͏ʹॲཧΛ
    هड़͢Δ
    •  ඇߏ଄Խσʔλʹର͢Δॊೈͳ
    ॲཧ͕ߦ͑Δ
    •  ߦࢦ޲ͷॲཧ
    
    %BUBGSBNF
    •  42-ϥΠΫʹॲཧΛهड़͢Δ
    •  εΩʔϚΛར༻ͨ͠ߏ଄Խσʔλॲ ཧ
    •  ྻࢦ޲ͷॲཧ
    
    
    %BUBGSBNF͸ɺ4QBSL͔Βొ৔ͨ͠ϋΠϨϕϧ"1*Ͱɺ
    ɾΦϓςΟϚΠβʹΑΔ࠷దԽͷԸܙΛड͚Δ
    ɾίʔυͷՄಡੑ্͕͕Δ
    ͱ͍ͬͨಛ௃͕͋Δ
    
14. 3%%ͱ%BUB'SBNF
    Spark Core Spark SQL Spark Streaming （ストリーム処理） Mllib (機械学習) GraphX (グラフ処理) SQL CSV {json} S3 BigQuery parquet Data Sources •  3%%͸4QBSL
    $PSFɺ%BUB'SBNF͸4QBSL42-ʹؚ·ΕΔػೳ
    •  4QBSLͷػೳ͸ঃʑʹ%BUB'SBNFϕʔεʹஔ͖׵Θ͍ͬͯΔ
      4USVDUVSFE
    4USFBNJOH4QBSL
    .-(SBQI
    'SBNFT
    RDD DataFrame
15. %BUB'SBNFͷΦϓςΟϚΠβʹΑΔ࠷దԽ
    •  ޮ཰తͳॲཧͷॱ൪ʹೖΕସ࣮͑ͯߦͯ͘͠ΕΔ
    •  σʔλιʔεʹΑͬͯ͸ɺϑΟϧλॲཧΛσʔλɾιʔεଆͰߦ͍ɺ ඞཁͳσʔλͷΈΛಡΈࠐΉΑ͏ʹ͢Δ
    hIps://databricks.com/blog/2015/03/24/spark-sql-graduates-from-alpha-in-spark-1-3.html
16. 3%%Ͱ8PSE$PVOUͯ͠ΈΑ͏ʂ
    Jupyter Notebookからも かんたんに使える！ 加工して 集計して 並び替え
17. %BUB'SBNFͰूܭॲཧΛ΍ͬͯΈΑ͏ʂ
    フィルターして グループごとに 集めて カウントする SQLライクにかける！
18. 1Z4QBSLͷΞʔΩςΫνϟ
    •  3%%
    ϫʔΧʔϊʔυͰͷॲཧ͸1ZUIPOϓϩηεͰߦΘΕΔ
    •  %BUB'SBNF
    ϫʔΧʔϊʔυͰͷॲཧ͸+7.্ͰߦΘΕΔ
    ͨͩ͠6%'͸1ZUIPOϓϩηεͰ࣮ߦ͞ΕΔ
    Master Worker Spark Context Java SparkContext Executor Task Task Python Python socket Py4J pipe Master Worker Spark Context Java SparkContext Executor Task Task socket Py4J
19. 1Z4QBSLͷΞʔΩςΫνϟ
    •  3%%
    ϫʔΧʔϊʔυͰͷॲཧ͸1ZUIPOϓϩηεͰߦΘΕΔ
    •  %BUB'SBNF
    ϫʔΧʔϊʔυͰͷॲཧ͸+7.্ͰߦΘΕΔ
    ͨͩ͠6%'͸1ZUIPOϓϩηεͰ࣮ߦ͞ΕΔ
    Master Worker Spark Context Java SparkContext Executor Task Task Python Python socket Py4J pipe Master Worker Spark Context Java SparkContext Executor Task Task socket Py4J ύϑΥʔϚϯε্ͷ໰୊఺
    •  *UFSBUPS୯ҐͰͷTFSJBMJ[BUJPOͱQZUIPOϓϩηε΁ͷύΠϓ ͕ൃੜ
    •  QJDLMJOHͱ+7.ͷೋॏͷTFSJBMJ[BUJPOʹΑΔίετ૿
    •  1ZUIPO
    8PSLFSͷىಈ
    •  1ZUIPO
    8PSLFSͷϝϞϦ͸+7.ͷ੍ޚର৅֎
    
20. ύϑΥʔϚϯεൺֱ
    出典：DataBricks社のブログ hIps://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data- science.html 3%%
    1ZUIPO͸஗͍ʂ
    
21. 1Z4QBSLͷ͍͍ͱ͜Ζʂ
    •  1Z%BUBϥΠϒϥϦͱ૊Έ߹Θͤͯ࢖͏͜ͱ͕Ͱ͖Δ
    –  3%%ͷதͰ/VNQZ΍4DJQZΛ࢖͏
    –  ूܭ݁ՌΛ1BOEBT
    %BUB'SBNFʹม׵ͯ͠.BUQMPUMJCͰ
    ՄࢹԽ͢Δ
    •  .-ϥΠϒϥϦ౳ʹ͍ͭͯ΋ɺ΄΅શͯར༻Ͱ͖Δ
    
22. 1Z4QBSLΤίγεςϜͷਐԽ
    •  1Z4QBSLͰ͸ɺQBOEBT
    %BUB'SBNFͱ4QBSL
    %BUB'SBNFͷ ૬ޓม׵͕Ͱ͖Δɻ
    –  QBOEBTͷEG͔ΒTQBSLͷEG
    •  TQBSLDSFBUF%BUB'SBNF QE@EG
    –  TQBSLͷEG͔ΒQBOEBTͷEG
    •  TQBSL@EGUP1BOEBT
    
    
    ͕ʂʂ্هͷม׵͸ڪΖ͘͠஗͍ɻɻɻ
    ʢಛʹલऀʣ
    
    •  ͜ΕΛղফ͢΂͘"QBDIF
    "SSPXͷ։ൃ͕ਐΜͰ͍Δ
    
23. "QBDIF
    "SSPX
    •  σʔλϑΥʔϚοτͷ࢓༷ͱͦΕΛར༻͢ΔͨΊͷϥΠϒϥϦ
    •  ҟͳΔݴޠɾϓϩμΫτؒͰͷσʔλ࿈ܞίετΛԼ͛Δ
    •  ։ൃதͷ4QBSLͰ
    TQBSLDPOGTFU lTQBSLTRMFYFDVUJPOBSSPXFOBCMFz
    lUSVFz ͱ͢Δ͜ͱͰɺ 4QBSL
    %BUB'SBNFΛUP1BOEBT ͢Δࡍʹɺ"SSPXͷϑΥʔϚοτ͕
    ࢖͑ΔΑ͏ʹͳΔ༧ఆ
    •  ଞʹ΋6%'ͷ7FDUPSJ[BUJPOͳͲ΋ܭը͞Ε͍ͯΔ
    出典：hIps://arrow.apache.org/
24. 4QBSL
    .BDIJOF
    -FBSOJOH
    •  4QBSLʹ͸ɺ3%%ϕʔεͷ.MMJCͱ%BUB'SBNFϕʔεͷ .-ͱ͍͏ͭͷػցֶशϥΠϒϥϦ͕͋Δɻ
    –  ঃʑʹ.-ʹد͍ͤͯΔ
    •  TDJLJUMFBSOͷӨڹΛڧ͘ड͚͍ͯΔ
    
25. 1Z4QBSLͰ࢖͑Δ୅දతͳػցֶशख๏Ұཡ
    •  $MBTTJpDBUJPO
    3FHSFTTJPO
    –  47.TɺϩδεςΟοΫճؼɺܾఆ໦ɺφΠʔϒϕΠζɺϥϯμϜϑΥϨε τɺઢܗճؼɺ(#5ɺϚϧνύʔηϓτϩϯɺJTPUPOJDճؼɺ"'5ੜଘճؼ
    •  $MVTUFSJOH
    –  ,NFBOTɺ-%"ɺ(..
    •  ڠௐϑΟϧλϦϯάɺύλʔϯϚΠχϯά
    –  "-4ɺ/.'ɺ'1(SPXUI
    •  ࣍ݩ࡟ݮ
    –  47%ɺ1$"
    •  'FBUVSF
    &YUSBDU
    BOE
    5SBOTGPSN
    –  5'*%'ɺ8PSE7FDɺ4UBOEBSE4DBMFSɺ/PSNBMJ[FSɺ/HSBNɺ 0OF)PU&ODPEFSɺ4USJOH*OEFYFSɺ-BCFMFE1PJOUɺ%$5
    •  #BTJD
    TUBUJTUJDT
    –  Χʔωϧີ౓ਪఆɺΧΠೋ৐ݕఆɺίϧϞΰϩϑεϛϊϑݕఆ
    •  &WBMVBUPS
    5VOJOH
    (FOFSBUPS
    –  "6$ɺ$SPTT
    7BMJEBUPSɺ1BSBN(SJE#VJMEFSɺ֤छσʔλ(FOFSBUPS
    FUDʜ
    
26. 4QBSLͰϨίϝϯσʔγϣϯ෼ΫοΩϯάʂ
    デモ
27. ʢ͍ͭͰʹ঺հʣ"QBDIF
    ;FQQFMJO
    •  +VQZUFS
    /PUFCPPLͱಉ͡Α͏ʹ࢖͑ΔՄࢹԽπʔϧ
    •  ༷ʑͳ࣮ߦΤϯδϯΛαϙʔτ͢Δ
    
28. ෼ੳݱ৔Ͱͷ1Z4QBSLͷ͔͍ͭͲ͜Ζ
    •  ,1*ϨϙʔςΟϯάͷΑ͏ͳ෼ੳͷݱ৔Ͱ͸ɺΘ͟Θ͟ େྔσʔλΛѻΘͣͱ΋αϯϓϦϯάͯ͠୆ͷαʔό ্ͰॲཧΛߦ͏΄͏͕ྑ͍৔߹΋ଟ͍ɻ
    –  ͨ͘͞ΜͷσʔλͰΫϥελϦϯάͳͲͷֶशΛߦ͏͜ͱͰ
    ͔͑ͬͯաֶशʹͳΔ͜ͱ΋͋Δɻ
    •  Ϩίϝϯσʔγϣϯɺҟৗݕ஌ɺ޿ࠂ഑৴࠷దԽ౳ɺ
    େن໛σʔλʹରͯ͠ػցֶशΛߦ͏ඞཁ͕͋Δ৔໘Ͱ ࢖ΘΕΔɻ
    ࢖͍ॴͷݟఆΊ͸͍ͩ͡ɻ
    
29. 3FUUZʹ͓͚Δ1Z4QBSLࣄྫ
    
30. 3FUUZͷ݄ؒສ66Λࢧ͑Δ෼ੳج൫
    ReIyαʔϏεج൫ ReIy෼ੳج൫ʢAWSʣ ReIy෼ੳج൫ʢGCPʣ Kinesis S3 EMR (Spark) S3 EC2 EC2 RDS(MySQL) 分析者 プランナ ここでSparkを 使っている
31. &5-ͷ֓ཁ
    •  3FUUZͷΞΫηεϩά
    –  ೔ʹ਺े(#ʢH[KTPOঢ়ଶʣͷϩά
    –  SFRVFTU63*΍VTFS
    BHFOUͳͲΛ෼ੳ͠΍͍͢ܗʹ੔ܗ
    –  ෼ੳ༻ͷTFTTJPO
    *%Λ&5-ͷաఔͰ෇༩
    •  TFDPOEBSZ
    TPSUͱNBQQBSUJUJPOΛ࢖ͬͨ
    
32. 1Z4QBSLΛ࠾༻ͨ͠ཧ༝
    •  ϚωʔδυɾαʔϏεʢ&.3ʣΛར༻Ͱ͖ΔͨΊɺ
    ؀ڥߏங͕ෆཁ
    –  EBTL΍$FMFSZ΋ީิͱͯ͠͸͋Δ͕ɺ؀ڥߏங͕ඞཁͱͳΔɻ ·ͨ͜ΕΒ͸σʔλྔͷεέʔϧʹ଱͑ΒΕͳ͍Մೳੑ͕͋Δ
    •  ࣾ಺ʹ1ZUIPO࢖͍͕ଟ͍ͨΊɺϝϯςφϯεΛߟྀͯ͠ 4DBMBͰͳ͘1ZUIPOΛ࠾༻ͨ͠
    •  )JWF΍1SFTUPʢ)BEPPQΤίγεςϜʣͱൺ΂ɺඇߏ଄ σʔλʹର͢Δॊೈͳදݱ͕΍Γ΍͍͢
    –  ෼ੳཁ݅ͷͨΊɺΞϓϦέʔγϣϯଆͰৼΒΕΔTFTTJPO*%ͱ͸ ผͰTFTTJPO*%૬౰ͷ΋ͷΛ෇༩ͯ͠΄͍͠ͱͷཁ๬͕͋Γɺ
    ͜ΕΛຬͨͨ͢ΊʹɺෳࡶͳίϨΫγϣϯॲཧΛߦ͏ඞཁ͕ ͋ͬͨ
    ちなみに、想定通り、ここの処理だけRDDなので遅い。 が、許容できる範囲だった。
33. 1Z4QBSL ʴ&.3 ͷύϑΥʔϚϯε্ͷ5JQTΛ͍͔ͭ͘ 
    •  ͳΔ΂͘%BUB'SBNFΛ࢖͏
    •  ͳΔ΂͘ॲཧ͢ΔσʔλྔΛݮΒ͢
    –  KPJO͸খ͍͞σʔλಉ࢜Λઌʹ
    •  ετϨʔδͷར༻
    –  ӬଓԽϑΝΠϧ͸4
    –  தؒϑΝΠϧ͸)%'4
    •  σʔλϑΥʔϚοτ͸1BSRVFUΛར༻͢Δͱߴ଎
    •  σʔλͷ4LFXʹ஫ҙ
    –  ҰͭͷύʔςΟγϣϯ͚ͩॲཧ͢Δσʔλྔ͕ଟ͘ͳͬͯɺͦ ͜ͷॲཧ଴ͪʹͳΔ
    
34. 1Z4QBSL ʴ&.3 ͷύϑΥʔϚϯε্ͷ5JQTΛ͍͔ͭ͘ 
    •  ϝϞϦνϡʔχϯάͷίπ
    –  1ZUIPOͰ4QBSLΛ࢖͏৔߹ɺ4DBMBͱൺ΂ͯΦϑώʔϓʢ+7.֎ ͷϝϞϦʣΛར༻͢ΔɻͦͷͨΊɺZBSOʹΑͬͯίϯςφ͕LJMM ͞ΕΔΤϥʔ͕ى͖΍͍͢ɻTQBSLQZUIPOXPSLFSNFNPSZ΍ TQBSLZBSOFYFDVUPSNFNPSZ0WFSIFBEύϥϝʔλͰνϡʔχϯ άΛߦ͏
    •  "QBDIF
    ;FQQFMJO
    –  &.3Ͱ͸ɺ+VQZUFS
    /PUFCPPLΛΠϯετʔϧ͠ͳͯ͘΋ɺ "QBDIF
    ;FQQFMJOͱ͍͏/PUFCPPL͕࢖͑Δɻ
    
35. ·ͱΊ
    •  4QBSL͸ɺେྔσʔλΛߴεϧʔϓοτͰॲཧ͢Δ͜ͱ ͷͰ͖ΔศརͳϓϩμΫτ
    •  ػցֶश΍ετϦʔϜͳͲ๛෋ͳػೳ͕͋Δ
    •  &.3΍%BUB1SPDͳͲͷϚωʔδυαʔϏε͕ศར
    •  1Z4QBSLΛར༻͢Δ৔߹͸Φʔόϔου͕͋Δ
    •  ࢖͍Ͳ͜ΖΛ͖ͪΜͱཧղ͢Δ͜ͱ͕͍ͩ͡
    •  3FUUZͰ͸σʔλΤϯδχΞϦϯάʹ1Z4QBSLΛ࠾༻ͯ͠ ͍Δ
    みなさんも、PySparkで大量データ処理やってみよう！
36. ͝ਗ਼ௌ͋Γ͕ͱ͏͍͟͝·ͨ͠J
    
37. ิ଍ 1ZUIPO੡ͷଞͷฒྻ෼ࢄॲཧܥͱͷൺֱ
    1ZUIPO੡ͷฒྻ෼ࢄॲཧܥʢDFMFSZ
    
    EBTLʣͱ4QBSLͷେ͖ ͳҧ͍ͱͯ͠͸ɺΫϥελن໛ͷΧόʔྖҬ͕͋͛ΒΕΔɻ
    ·ͨɺ4QBSLͷ৔߹ɺΫϥ΢υΛ࢖͏͜ͱͰ؀ڥߏஙΛߦ Θͳͯ͘ྑ͍ͱݴ͏ϝϦοτ͕͋Δɻ
    サーバ台数の目安 Celery dask (py)spark 1台 4〜5台 数千台
38. ิ଍ 4QBSLϢʔβͷ֤ݴޠར༻ऀͷׂ߹
    出典：Spark Survery 2016 hIps://databricks.com/blog/2016/09/27/spark-survey-2016-released.html
39. ิ଍ ηΧϯμϦιʔτͱNBQQBSUJUJPOͷઆ໌
    [(0,4), (2,4), (3,1), (2,2), (0,1), (1,1),(1,8), (3,6)] [(0,1), (0,4), (2,2), (2,4)] [(0,1), (1,1), (1,8), (3,1), (3,6)] もとの配列 パーティションキー とそれ以外の値で ソートする [(0,1), (0,5), (2,2), (2,5)] [(0,1), (1,1), (1,9), (3,1), (3,7)] secondary sort mappar??on par??onごとに一度 処理を実行する