存储过程实现上亿级图数据分块ETL

apoc.load.jdbc

 olab.ids.batch

apoc.periodic.iterate

olab.escape

WITH 'jdbc:mysql://datalab-contentdb-dev.crkldnwly6ki.rds.cn-north-1.amazonaws.com.cn:3306/database?user=dev&password=datalabgogo&useUnicode=true&characterEncoding=utf8&serverTimezone=UTC' AS url,'SELECT MIN(puid) AS min,MAX(puid) AS max FROM table' AS sql,10000000 AS batch
CALL apoc.load.jdbc(url,sql) YIELD row WITH row.min AS min,row.max AS max,batch
WITH olab.ids.batch(min,max,batch) AS value
UNWIND value AS list
RETURN list[0] AS min,list[1] AS max

https://github.com/ongdb-contrib/ongdb-lab-apoc

https://github.com/neo4j-contrib/neo4j-apoc-procedures

ongdb-3.5.x 
apoc-3.5.x 
olab-apoc-3.5.x

WITH 'jdbc:mysql://datalab-contentdb-dev.crkldnwly6ki.rds.cn-north-1.amazonaws.com.cn:3306/database?user=dev&password=datalabgogo&useUnicode=true&characterEncoding=utf8&serverTimezone=UTC' AS url,'SELECT MIN(puid) AS min,MAX(puid) AS max FROM table' AS sql,1000000 AS batch,'SELECT parent_pcode AS `name`,CONVERT(DATE_FORMAT(hupdatetime,\'%Y%m%d%H%i%S\'),UNSIGNED INTEGER) AS hupdatetime FROM table WHERE hisvalid=1 AND parent_pcode IS NOT NULL AND puid>=? AND puid<=?' AS loadSql
CALL apoc.load.jdbc(url,sql) YIELD row WITH row.min AS min,row.max AS max,url,batch,loadSql
WITH olab.ids.batch(min,max,batch) AS value,url,batch,loadSql
UNWIND value AS list
WITH list[0] AS min,list[1] AS max,url,loadSql,'CALL apoc.load.jdbc({url},{loadSql},[{min},{max}])' AS jdbc
CALL apoc.periodic.iterate(olab.replace(jdbc,[{raw:'{url}',rep:'\''+url+'\''},{raw:'{loadSql}',rep:'\''+olab.escape(loadSql)+'\''},{raw:'{min}',rep:min},{raw:'{max}',rep:max}]),'MERGE (n:PREPCODE {name:row.name}) SET n+=row',{parallel:false,batchSize:1000,iterateList: false}) YIELD batches,total,timeTaken,committedOperations,failedOperations,failedBatches,retries,errorMessages,batch,operations RETURN batches,total,timeTaken,committedOperations,failedOperations,failedBatches,retries,errorMessages,batch,operations;