

エンジニア4年目(python未経験)なのですが  
業務でpythonを用いてスクリプトを開発しなければならず,自分なりに調べて下記の様なスクリプトを組んでみたのですがパフォーマンスが著しく悪く,そのためチューニングをしたいのですが,まだまだ勉強し始めたばかりでチューニング方針がなかなか決めれないでいます。。。  
納期も近いためこちらにてご相談させていただきました。  
どなたかどんな些細なことでもよいのでチューニングに関してご教示いただけますと幸いです。。  
特にスクリプトのの2重for文の部分をチューニングしたいです。

[code]

    import pandas as pd
    import numpy as np
    
    from pyspark import SparkConf,SparkContext
    
    from pyspark.sql import SparkSession
    from pyspark.shell import SQLContext
    import time
    
    spark = SparkSession.builder.master("yarn").config(conf=SparkConf()).getOrCreate()
    sc = SparkContext.getOrCreate()
    sqlContext = SQLContext(sc)
    
    spark.conf.set("spark.sql.execution.arrow.enabled", "true")
    
    
    spark = SparkSession.builder \
            .appName('Spark SQL and DataFrame') \
            .getOrCreate()
    
    #S3のファイル格納パス
    filepass_aaa = 's3://*******/******/test/aaa.csv'
    filepass_bbb = 's3://*******/******/test/bbb.csv'
    
    #CSVファイルをS3から読み込みSparkのDataFreameを生成
    sdf_aaa = sqlContext.read.format("com.databricks.spark.csv").option("header","true").load(filepass_aaa)
    sdf_bbb = sqlContext.read.format("com.databricks.spark.csv").option("header","true").load(filepass_bbb)
    
    #pandasのDataFreameの型に変更
    pdf_aaa = sdf_aaa.toPandas()
    pdf_bbb = sdf_bbb.toPandas()
    
    条件に合致した場合にaaaの該当の値をbbbの値で置換
    def mapping(row,row2):
        pdf_aaa.at[(pdf_aaa['ID'] == row[0]),'○○ID'] = row2[2]
        pdf_aaa.at[(pdf_aaa['ID'] == row[0]),'×××'] = row2[1]
        pdf_aaa.at[(pdf_aaa['ID'] == row[0]),'△△△'] = row2[3]
        pdf_aaa.at[(pdf_aaa['ID'] == row[0]),'□□□'] = row2[0]
        pdf_aaa.at[(pdf_aaa['ID'] == row[0]),'■■■'] = row2[4]
        pdf_aaa.at[(pdf_aaa['ID'] == row[0]),'◇◇◇'] = row2[5]
    
    
    for index, row in pdf_aaa.iterrows():
    
        for index2,row2 in pdf_bbb.iterrows():
    
            #条件：名称+住所+TEL
            if(row[7] == row2[6]) & (row[9] == row2[8]) & (row[8] == row2[7]):
                mapping(row,row2)
                break
            #条件：名称+TEL
            elif(row[7] == row2[6]) & (row[8] == row2[7]):
                mapping(row,row2)
                break
            #条件：名称+住所
            elif(row[7] == row2[6]) & (row[9] == row2[8]):
                mapping(row,row2)
                break
            #条件：TEL+住所
            elif(row[8] == row2[7]) & (row[9] == row2[8]):
                mapping(row,row2)
                break
            #条件：住所+URL
            elif(row[9] == row2[8]) & (row[10] == row2[9]):
                mapping(row,row2)
                break
            #条件：TEL+URL
            elif(row[8] == row2[7]) & (row[10] == row2[9]):
                mapping(row,row2)
                break
            #条件：TEL
            elif(row[8] == row2[7]):
                mapping(row,row2)
                break
            else:
                continue
                break
    
    pdf_out = pdf_aaa[~pdf_aaa['○○ID'].isnull()]
    pdf_out_null = pdf_aaa[pdf_aaa['○○ID'].isnull()]
    
    pdf_out.to_csv('result.csv',header=True, index=False)
    pdf_out_null.to_csv('result_null.csv',header=True, index=False)
    
[/code]

【実現したいこと】  
DataFrame aaaのデータをDataFrame bbbのデータと比較し特定の条件に合致した場合のみ特定のaaaのデータをbbbのデータで置換する  
aaa,bbbともにデータは約100万件ほど


CC BY-SA 4.0
2019-03-07T11:58:32.910
53278
2019-03-08T12:49:56.840
2019-03-07T12:59:23.950
32451
32451
question
python3 pandas spark
python,pandasでのパフォーマンスチューニング


順番に

(1)  
まず、この手の質問であれば、コードの前半を占めるデータを取得する部分は全く不要ですので省いていただいて構いません。  
が、その代わりに各DataFrameがどのようなデータ構成なのかは必ず提示したください。できましたらダミーで構いませんので実際にコードを動作させることができるデータを提示していただけると助かります。  
今回はとりあえず、

**【pdf_aaa の構成】**

[code]

    No|Column名| 内容
    0 |ID      |ID番号
    1 |□□□     |何かしらの情報
    2 |×××     |何かしらの情報
    3 |○○ID    |何かしらの情報
    4 |△△△   |何かしらの情報
    5 |■■■     |何かしらの情報
    6 |◇◇◇   |何かしらの情報
    7 |Name    |名前
    8 |Tel     |電話番号
    9 |Addr    |住所
    10|URL     |URL
    
[/code]

**【pdf_bbb の構成】**

[code]

    No|Column名| 内容
    0 |□□□     |□□□ の上書情報
    1 |×××     |××× の上書情報
    2 |○○ID    |○○ID の上書情報
    3 |△△△   |△△△ の上書情報
    4 |■■■     |■■■ の上書情報
    5 |◇◇◇   |◇◇◇ の上書情報
    6 |Name    |名前
    7 |Tel     |電話番号
    8 |Addr    |住所
    9 |URL     |URL
    
[/code]

と仮定して回等します

(2)  
現状のコードでは

[code]

    elif(row[8] == row2[7]):
    
[/code]

のようにColumn番号で記述している箇所が多いようですが、できましたら適切な Calumn名を設定して

[code]

    elif(row['Tel'] == row2['Tel']):
    
[/code]

のように記述したほうが、なんの処理を行っているのかが分かり易くなり、保守性がよくなるかと思います。

> 以下、(1)で記述した Column名を使って記述します。

(3)  
`mapping()`関数において

[code]

    def mapping(row,row2):
        pdf_aaa.at[(pdf_aaa['ID'] == row[0]),'○○ID'] = row2[2]
        ...
    
[/code]

のように pdf_aaaよりID列で再度行を選択しておりますが、ループにて既にpdf_aaaの **Index** 値が得られておりますので

[code]

    def mapping(idx,row2):
        pdf_aaa.at[idx,'○○ID'] = row2[2]
        ...
    
[/code]

で良いのではないでしょうか。

> この際に、呼び出し側は`mapping(index,row2)`となります。

更には ６箇所の更新をまとめて

[code]

    def mapping(idx,row2):
        pdf_aaa.loc[idx,'□□□':'◇◇◇'] = row2['□□□':'◇◇◇'].values
    
[/code]

のようにも記述できるかと思います。

(4)  
Pandasにおいてループはかなり遅い処理となりますので、できれば避けた方が無難です。  
とりあえず内側のループを排除する方法を考えます。PandasのFilteringを使うと１行ずつ比較する意味はありませんので、

[code]

    for index, row in pdf_aaa.iterrows():
    
        #条件：名称+住所+TEL
        # pdf_bbbを上記の条件でフィルタリング
        row2 = pdf_bbb.loc[(pdf_bbb['Name'] == row['Name']) &
                           (pdf_bbb['Addr'] == row['Addr']) &
                           (pdf_bbb['Tel'] == row['Tel'])]
        if len(row2) > 0:
            # 複数行選択されることを考慮して、１行めだけ渡す
            mapping(index, row2.iloc[0])
            break
    
[/code]

のように書けるかと思います。

pandas.merge()などで最終的に外側のループもとることもできると思いますが、構成が全く違うものとなりますので、とりあえずはここまで。


CC BY-SA 4.0
2019-03-08T12:36:44.390
53299
2019-03-08T12:49:56.840
2019-03-08T12:49:56.840
24801
24801
53278
answer