シェープファイルの属性をaccessに高速で登録したい。

実現したいこと

調べたこと

https://teratail.com/questions/u88jge5g5b2qdx
https://qiita.com/ku_a_i/items/77f4680651e628517db9
https://stackoverflow.com/questions/69451791/error-using-to-sql-method-multi-with-ms-access
https://github.com/gordthompson/sqlalchemy-access/wiki/%5Bpandas%5D-faster-alternative-to-.to_sql()-for-large-uploads

つまり
https://teratail.com/questions/326746
の質問文で記載したようにGetFieldDefnから型を取得して対応するAccessの型を判定し
それにより"ADODB.Connection"使ってテーブルを作成して
そこにinsert-intoで登録すれば良い。という事になるのでしょうか。

こんな方法しかないのでしょうか。
他に良い方法等ありましたら教えて下さい。

とりあえず、to_sqlで出力する例1とADOつかってRecorsSetでappendする例2と
リンク先に提示してあったto_excelからのinsert intoの例3を比較してみました。

コードは以下の通りです。

python
1import pandas as pd
2from urllib.parse import quote_plus
3from sqlalchemy import create_engine
4from sqlalchemy import types
5import time
6import os
7import math
8
9def alchemy_engine(db_path):
10    con_str = "DRIVER=" + \
11              "{Microsoft Access Driver (*.mdb, *.accdb)};" + \
12               f"DBQ={db_path};"
13    con_str = quote_plus(con_str)
14    engine = create_engine(
15        f"access+pyodbc:///?odbc_connect={con_str}",
16        echo=True)
17 
18    return engine
19
20path_access=r"D:\work\study\python\pandas_access\orderby.accdb"
21engine = alchemy_engine(path_access)
22
23data_a=[]
24data_b=[]
25data_c=[]
26data_d=[]
27data_e=[]
28for i in range(30000):
29    data_a.append(i)
30    data_b.append(float(i))
31    data_c.append(str(i))
32    data_d.append(None)
33    data_e.append(str(i))
34
35df=pd.DataFrame({'A':data_a,'B':data_b,'C':data_c,'D':data_d,'E':data_e})
36dtypes={'A':types.Integer,'B':types.Float,'C':types.String(10),'D':types.DateTime,'E':types.Text}
37
38start_time = time.perf_counter()
39df.to_sql('testtable', engine, if_exists='replace', index=False,dtype=dtypes)
40end_time = time.perf_counter()
41 
42elapsed_time = end_time - start_time
43print(elapsed_time)
44
45import win32com.client
46
47
48def insert_to_access(path,df,table,dtypes):
49    dic_data_type = {
50        types.DateTime:"DATETIME",
51        types.Float:"DOUBLE",
52        types.Integer:"INTEGER",
53        types.Text: "LONGCHAR",
54        types.String:"VARCHAR",
55    }
56    conn = win32com.client.Dispatch(r'ADODB.Connection')
57
58    DSN ='Provider=Microsoft.ACE.OLEDB.12.0;Data Source=' + path + ';'
59
60    conn.Open(DSN)
61            
62    sql="DROP TABLE [" + table + "]"
63    try:
64        conn.Execute(sql)
65    except:
66        pass
67
68    sql="CREATE TABLE [" + table + "] ("
69    for column in df.columns:
70        if isinstance(dtypes[column], types.String):
71           sql += column + ' ' +  dic_data_type[types.String]
72           sql += '(' + str(dtypes[column].length) + ')'
73        else:
74           sql += column + ' ' +  dic_data_type[dtypes[column]]
75        sql += ','
76    sql=sql[:-1]+ ')'
77
78    conn.Execute(sql)
79    
80    rs = win32com.client.Dispatch(r'ADODB.Recordset')
81    rs.CursorLocation = 3 # adUseClient
82    rs.LockType = 4 # adLockBatchOptimistic 
83    rs.Properties.Item("Append-Only Rowset").Value= True
84    rs.Open("SELECT * FROM [" + table + "]",conn)
85
86    l=len(df.columns)
87    count=0
88    for row in df.itertuples(name=None):
89        count+=1
90        rs.AddNew()
91        for i in range(l):
92            if not row[i+1] is None:
93                if isinstance(row[i+1],float) or isinstance(row[i+1],int):
94                    if not math.isnan(row[i+1]):
95                        rs.Fields.Item(i).Value=row[i+1]
96                else:    
97                    rs.Fields.Item(i).Value=row[i+1]
98        if count % 5000 ==0:  
99            rs.UpdateBatch()
100    if count % 5000 != 0:  
101        rs.UpdateBatch()
102    rs.Close()
103    conn.Close()
104
105def insert_to_access2(path,df,table,dtypes):
106    dic_data_type = {
107        types.DateTime:"DATETIME",
108        types.Float:"DOUBLE",
109        types.Integer:"INTEGER",
110        types.Text: "LONGCHAR",
111        types.String:"VARCHAR",
112    }
113    conn = win32com.client.Dispatch(r'ADODB.Connection')
114    DSN ='Provider=Microsoft.ACE.OLEDB.12.0;Data Source=' + path + ';'
115    conn.Open(DSN)
116            
117    sql="DROP TABLE [" + table + "]"
118    try:
119        conn.Execute(sql)
120    except:
121        pass
122
123    sql="CREATE TABLE [" + table + "] ("
124    for column in df.columns:
125        if isinstance(dtypes[column], types.String):
126           sql += column + ' ' +  dic_data_type[types.String]
127           sql += '(' + str(dtypes[column].length) + ')'
128        else:
129           sql += column + ' ' +  dic_data_type[dtypes[column]]
130        sql += ','
131    sql=sql[:-1]+ ')'
132
133    conn.Execute(sql)
134    
135    xlsx_path = os.path.splitext(path)[0] + '.xlsx'
136    df.to_excel(xlsx_path, index=False)
137    sql = f"""\
138    INSERT INTO [{table}]
139    SELECT * FROM [Sheet1$] IN "{xlsx_path}" 'Excel 12.0 Macro;HDR=Yes'
140    """
141
142    conn.Execute(sql)
143    os.remove(xlsx_path)
144    conn.Close()
145
146
147start_time = time.perf_counter()
148insert_to_access(path_access,df,'testtable2',dtypes)
149end_time = time.perf_counter() 
150elapsed_time = end_time - start_time
151print(elapsed_time)
152
153
154start_time = time.perf_counter()
155insert_to_access2(path_access,df,'testtable3',dtypes)
156end_time = time.perf_counter() 
157elapsed_time = end_time - start_time
158print(elapsed_time)
159

結果は
例1が22秒
例2が35秒(やってみたがto_sqlより遅いとは…何かまずい所があるのかな？)
例3が6秒
でした。

やはり、リンク先の内容の通りということなのでしょうか。
データの量を増やしてどう変わるかについては、また検証してみます。

あとシェープファイルからということでしたので
シェープファイルからの登録の例として書いてみたコードも記載しておきます。

python
1import osgeo.ogr as ogr
2from sqlalchemy.types import Float,Integer,String,Text,DateTime
3
4path_read_shp=r"D:\study\Python\AAAA.shp"
5
6driver = ogr.GetDriverByName("ESRI Shapefile")
7data_source = driver.Open(path_read_shp)
8layer = data_source.GetLayer(0)
9layer_defn = layer.GetLayerDefn()
10dtypes={}
11for column in range(layer_defn.GetFieldCount()):
12    field_defn=layer_defn.GetFieldDefn(column)
13    field_name=field_defn.GetName()    
14    if field_defn.GetType() == ogr.OFTInteger:
15        dtypes[field_name]=Integer
16    elif field_defn.GetType() == ogr.OFTInteger64:
17        dtypes[field_name]=Integer
18    elif field_defn.GetType() == ogr.OFTReal:
19        dtypes[field_name]=Float
20    elif field_defn.GetType() == ogr.OFTString:
21        if field_defn.GetWidth()<=255:
22            dtypes[field_name]=String(field_defn.GetWidth())
23        else:
24            dtypes[field_name]=Text
25    elif field_defn.GetType() == ogr.OFTDate :
26        dtypes[field_name]=DateTime
27
28data_source.Destroy()
29
30import pandas as pd
31import geopandas as gpd
32from urllib.parse import quote_plus
33from sqlalchemy import create_engine
34
35# sqlalchemyのengineを作成
36def alchemy_engine(db_path):
37    con_str = "DRIVER=" + \
38              "{Microsoft Access Driver (*.mdb, *.accdb)};" + \
39               f"DBQ={db_path};"
40    con_str = quote_plus(con_str)
41    engine = create_engine(
42        f"access+pyodbc:///?odbc_connect={con_str}",
43        echo=True)
44 
45    return engine
46
47engine = alchemy_engine(r"D:\study\Python\sss.accdb")
48
49gdf=gpd.read_file(path_read_shp,encode='utf-8')
50df=pd.DataFrame(gdf.drop('geometry',axis=1))
51df.to_sql('shape_test', engine, if_exists='replace', index=False)
52