ããã«ã¡ã¯ã ãšã³ã¿ãŒãã©ã€ãº ç¬¬äžæ¬éš ããŒã±ãã£ã³ã° ITéšã®çåã§ãã ãã®ããã°ã§ã¯ã é«éã«åäœãã忣åŠçãšã³ãžã³ã Apache Sparkã ãš ãªãŒãã³ããŒãã«ãã©ãŒããããDelta Lakeã ãåºç€ãšããã¬ã€ã¯ããŠã¹ç°å¢ãæ§ç¯ã§ããDatabricksäžã§ç®¡çããŠããã㌠ã¿ã»ãã ã«å¯ŸããŠã åå¯ã åŠçãè¡ãã¢ãããŒãã«ã€ããŠç޹ä»ããŸãã å®éã®ããŒãããã¯ã®åŠçã«ã€ããŠã玹ä»ããããšæã£ãŠããŸãããæ³å®ãããå
容ãå€ããªã£ãŠããŸã£ãã®ã§ã åå¯ã ã®æŠèŠã玹ä»ãããæŠèŠç·šãã ãœãŒã¹ã³ãŒã ãªã©å
·äœç㪠åå¯ã åŠçã®å
·äœçãªå
容ã玹ä»ããã æ±ºå®è« çãããã³ã°ç·šãã確ççãããã³ã°ç·šãã®äžéšäœã«ããããšæããŸãã Databricksã§å®çŸããããŒã¿åå¯ããæŠèŠç·šã Databricksã§å®çŸããããŒã¿åå¯ããæ±ºå®è«çãããã³ã°ç·šã Databricksã§å®çŸããããŒã¿åå¯ãã確ççãããã³ã°ç·šãïŒæ¬èšäºïŒ æ¬èšäºã¯ç¢ºççãããã³ã°ç·šãšãªããŸãã åå¯ã åŠçã®æŠèŠã«ã€ããŠèª¬æããããã°èšäºãå
¬éããŠãããŸãã®ã§ã åå¯ã åŠçã«ãŸã 銎æã¿ã®ãªãæ¹ã¯ããã²ãã¡ããã芧ããã ãããšå¹žãã§ãã æ¬èšäºã§ã¯Zinggãšããã©ã€ãã©ãªãå©çšãã確ççãããã³ã°ãè¡ãå
·äœçãªåŠçå
å®¹ãæµãã«ã€ããŠèª¬æããŸããZinggã®æŠèŠã«ã€ããŠã¯ãDatabricksã§å®çŸããããŒã¿ åå¯ã ãæŠèŠç·šããã§ç޹ä»ããŠãããŸãã®ã§ããã¡ããåãããŠã芧ãã ããã ãŸããæ¬èšäºã®å
容ã¯Zinggãå
¬éããŠããExampleãšDatabricksãå
¬éããŠãããœãªã¥ãŒã·ã§ã³ ã¢ã¯ã»ã© ã¬ãŒã¿ãåèã«ããŠãããŸãïŒExample + ãœãªã¥ãŒã·ã§ã³ ã¢ã¯ã»ã© ã¬ãŒã¿ãŒ + èªåã®æ¹ã§Databricksã®èšèšã«åãããŠå ç倿Žãå ãããããªåœ¢ïŒ é©å®ãã¡ãã®ããŒãžãåèã«ããŠããã ããªããæ¬èšäºã®æ¹ãèªãã§ããã ãããšè¯ãããšæããŸãã Zingg https://github.com/zinggAI/zingg/blob/main/examples/databricks/FebrlExample.ipynb https://www.zingg.ai/documentation-article/identity-resolution-on-databricks-for-customer-360 Databricks https://notebooks.databricks.com/notebooks/RCG/product-er-with-images/index.html#product-er-with-images_1.html ã§ã¯ãæ©éããŒã¿ã® åå¯ã ãéå§ããŠãããŸãããïŒ Zinggãå©çšããåå¯ãåŠçã®æŠèŠ 1. ã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã 1. ã¯ã©ã¹ã¿ãŒã«ã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã 1. Zinggã®JARãã¡ã€ã«ãGitHubããããŠã³ããŒãããŸãã 2. é©åœãªVolumeã«JARãã¡ã€ã«ãé
眮ãã 3. ã¯ã©ã¹ã¿ãŒã®èšå®ããã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã 2. ããŒãããã¯ã§Zinggã®Pythonããã±ãŒãžãã€ã³ã¹ããŒã«ãã 2. åå¯ãåŠçã«å¿
èŠãªã¢ãžã¥ãŒã«ãã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã ã©ã€ãã©ãªã®ã€ã³ã¹ããŒã« Zinggã§å©çšããã¯ãŒã¯ã¹ããŒã¹ãæŽç åå¯ãåŠçã§å©çšãã颿°ã®å®çŸ©ãã¢ãžã¥ãŒã«ã®import 3. Zinggã®èšå®ïŒConfigïŒãå®çŸ© 4. Zinggã®ã€ã³ããããã¢ãŠããããå
ãèšå®ãã 5. ãã£ãŒã«ãã®å±æ§ãæå®ãã 6. ãã©ã¡ãŒã¿ãèšå®ãã 7. ãã¬ãŒãã³ã°ããŒã¿äœæãžã§ããå®è¡ãã 8. ã©ããªã³ã°ã«äœ¿çšããããŒã¿ã»ãããäœæãã 9. ã©ããªã³ã°ãè¡ã 10. ã¢ãã«ã®ãã¬ãŒãã³ã° 11. ãããã³ã°çµæã確èªãã ãŸãšã Zinggãå©çšãã åå¯ã åŠçã®æŠèŠ Zinggãå©çšãã åå¯ã ã¯æ¬¡ã®ã¹ãããã§è¡ããŸãã ãŸãæåã« åå¯ã 察象ã㌠ã¿ã»ãã ã®é
ç®ã®å±æ§ïŒããã¹ããæ°å€é
ç®ãªã®ããã©ã®ãããªãããã³ã°ããããããªã©ïŒãããããå®çŸ©ããŸã å®çŸ©ããå
容ãå
ã«ã㌠ã¿ã»ãã ããäžèŽããŠããå¯èœæ§ã®é«ããã¢ããµã³ããªã³ã°ããŸã ãµã³ããªã³ã°ããããã¢åè£ãå
ã«ããŠãŒã¶ããã¢ã®äžèŽãäžäžèŽãäžæã倿ããã©ããªã³ã°ãè¡ããŸã ã¢ãã«ã®ã ã¬ãŒãã³ ã°ã«ååãªæ°ãéãŸããŸã§ãµã³ããªã³ã°ãšã©ããªã³ã°ãç¹°ãè¿ããŸã æåž«ããŒã¿ãå
ã«ã¢ãã«ãäœæãã ã¯ã©ã¹ã¿ ãŒåãè¡ããŸã ã¯ã©ã¹ã¿ ãŒåãããã㌠ã¿ã»ãã ã§ã¯ãã¢ã®äžèŽç¢ºçã確èªã§ããŸã äžèŽç¢ºçã確èªã§ããã®ã§ããXX%以äžã¯äžèŽãXX ~ YY%ã¯ãŠãŒã¶ã§æ¹ããŠæåãããã³ã°ãYY%以äžã¯å¥äººãšããããšãã£ããããªãªãã¬ãŒã·ã§ã³ãå¯èœã§ã ã§ã¯ã次ããå®éã«Databricksã§Zinggã® åå¯ã åŠçãè¡ãæ¹æ³ã«ã€ããŠãå
·äœçã«ç޹ä»ããŠãããŸãã 1. ã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã ZinggãããŒãããã¯ã§å©çšããããã«ã¯ã次ã®2ã¹ããããå¿
èŠã«ãªããŸãã ïŒããŒãããã¯ã«ã¢ã¿ããããïŒ ã¯ã©ã¹ã¿ ãŒã«ã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã ããŒãããã¯ã§zinggã® Python ããã±ãŒãžãã€ã³ã¹ããŒã«ãã æé ãé çªã«ç޹ä»ããŸãã 1. ã¯ã©ã¹ã¿ ãŒã«ã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã 1. Zinggã®JARãã¡ã€ã«ã GitHub ããããŠã³ããŒãããŸãã 以äžã® GitHub ã® ãªããžã㪠ãããææ°ã®Zingg JARãã¡ã€ã«ãããŠã³ããŒãããŸããïŒ25/5æçŸåš0.4.0ãææ°ïŒ https://github.com/zinggAI/zingg/releases 2. é©åœãªVolumeã«JARãã¡ã€ã«ãé
眮ãã ããŠã³ããŒãããJARãã¡ã€ã«ããä»»æã®Volumeã«ã¢ããããŒãããŸãã ã¯ã©ã¹ã¿ ãŒã®ã©ã€ãã©ãªã¯ ã¯ãŒã¯ã¹ããŒã¹ ãVolumeãS3ãªã©ã® ã¯ã©ãŠã ã¹ãã¬ãŒãžçã«é
眮ãããŠããå¿
èŠãããã®ã§ãVolume以å€ã®å Žæãè¯ãå Žåã¯é©å®é
眮å
ã倿ŽããŠãã ããã https://docs.databricks.com/aws/en/libraries/cluster-libraries 3. ã¯ã©ã¹ã¿ ãŒã®èšå®ããã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã ã¯ã©ã¹ã¿ ãŒèšå®ããŒãžãéãããã©ã€ãã©ãªãã¿ããããæ°èŠãã€ã³ã¹ããŒã«ããéžæããã¹ããã2ã§Volumeã«é
眮ããJARãã¡ã€ã«ãæå®ããŠã€ã³ã¹ããŒã«ããŸãã 2. ããŒãããã¯ã§Zinggã® Python ããã±ãŒãžãã€ã³ã¹ããŒã«ãã 1. ã§ã€ã³ã¹ããŒã«ãã ã¯ã©ã¹ã¿ ãŒãã¢ã¿ããããããŒãããã¯äžã§ã以äžã®ã»ã«ãå®è¡ããZinggã® Python ããã±ãŒãžãã€ã³ã¹ããŒã«ããŸãã %pip install zingg dbutils.library.restartPython() 2. åå¯ã åŠçã«å¿
èŠãªã¢ãžã¥ãŒã«ãã©ã€ãã©ãªãã€ã³ã¹ããŒã«ãã Zinggã®ä»ã«ãããŒãããã¯ã§äœ¿çšãããã®ä»ã®ã©ã€ãã©ãªãã¢ãžã¥ãŒã«ã倿°ããããããã€ã³ã¹ããŒã«ãŸãã¯æºåããŠãããŸãã ã©ã€ãã©ãªã®ã€ã³ã¹ããŒã« ããŒã¿ã衚圢åŒã§æŽåœ¢ããŠè¡šç€ºããããã® Python ããã±ãŒãž tabulate ãã€ã³ã¹ããŒã«ããŸã !pip install tabulate dbutils.library.restartPython() Zinggã§å©çšãã ã¯ãŒã¯ã¹ããŒã¹ ãæŽç Zinggã§ åå¯ã 察象ãšããã㌠ã¿ã»ãã ã®æ ŒçŽå ŽæããåŠççµæã®åºåå
ãšãªããã¹ãæå®ããŸãã ãã ããããã«é¢ããŠå°ã泚æãå¿
èŠã§ãZingg㯠Apache Sparkã§åäœãã åå¯ã ã©ã€ãã©ãªã§ãããDatabricksãšãã€ãã£ãã«çµ±åãããŠããããã§ã¯ãªããããå®è£
ããéã«æ°ãã€ããªããã°ãããªããã€ã³ãããããŸãã å
·äœçã«ã¯æ¬¡ã®ãããªç¹ã«æ³šæãå¿
èŠã§ãã Zinggã§ åå¯ã ãããã㌠ã¿ã»ãã ã¯Databricksäžã§ã¯å€éšããŒãã«ãšããŠç»é²ãããŠããå¿
èŠããã Zinggã§äœæãããã㌠ã¿ã»ãã ããã ã¬ãŒãã³ ã°ããŒã¿ãã¢ãã«ãªã©ã¯Unity Catalogã®ããŒãã«ãšããŠç»é²ããããšãã§ããªããããDBFSãããªã¥ãŒã ãªã©ã«äœæããã ZinggãšDatabricksã ã¯ã©ãŠã ã¹ãã¬ãŒãžã®é¢ä¿ããŸãšãããšæ¬¡ã®ãããªå³ã«ãªããŸãã ãã®ãããäºåã« ãã€ã³ããããããã㌠ã¿ã»ãã ãå€éšããŒãã«ãšããŠäœæããããZinggã®åºåå
ãšãªãVolumeãäœæãããŠããå¿
èŠããããŸãã æ¬èšäºã§ã¯äœææ¹æ³ã«ã€ããŠã¯å²æãããŠããã ããŸããã以äžããŒãžãåèã«äœæããŠããã ãããšè¯ãããšæããŸãã https://docs.databricks.com/aws/ja/tables/external https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-volumes ããŒãããã¯ã§ã¯æ¬¡ã®ããã«å€æ°ã«ãã¹ãæ ŒçŽããŠãããŸãã # Zinggã§äœæãããããŒã¿ã»ãããããã¬ãŒãã³ã°ããŒã¿ãã¢ãã«ãæ ŒçŽãããVolume # catalog, schema, volumeã¯ããããé©åœãªå€ãæå®ããŠãã ããã zinggDir = "/Volumes/catalog/schema/volume" # ã¢ãã«ã®IDãåå¯ãã®åŠçããšã«ç·šéããŠãã ããã modelId = "hogehoge-model" # åŸè¿°ã®åå¯ãåŠçã§ã©ãã«ä»ãããæ
å ±ãæ ŒçŽããããã©ã«ã MARKED_DIR = zinggDir + "/" + modelId + "/trainingData/marked/" UNMARKED_DIR = zinggDir + "/" + modelId + "/trainingData/unmarked/" åå¯ã åŠçã§å©çšãã颿°ã®å®çŸ©ãã¢ãžã¥ãŒã«ã®import åŸç¶ã®åŠçã§å©çšãã颿°ãå®çŸ©ããããã¢ãžã¥ãŒã«ãå
ã«importããŸãã import pandas as pd import numpy as np import time import uuid from tabulate import tabulate from ipywidgets import widgets, interact, GridspecLayout import base64 import pyspark.sql.functions as fn ##this code sets up the Zingg Python interface from zingg.client import * from zingg.pipes import * def cleanModel (isFull= False ): if (isFull): dbutils.fs.rm(zinggDir + "/" + modelId, recurse= True ) else : dbutils.fs.rm(MARKED_DIR, recurse= True ) # drop unmarked data dbutils.fs.rm(UNMARKED_DIR, recurse= True ) return # assign label to candidate pair def assign_label (candidate_pairs_pd, z_cluster, label): ''' The purpose of this function is to assign a label to a candidate pair identified by its z_cluster value. Valid labels include: 0 - not matched 1 - matched 2 - uncertain ''' # assign label candidate_pairs_pd.loc[ candidate_pairs_pd[ 'z_cluster' ]==z_cluster, 'z_isMatch' ] = label return def count_labeled_pairs (marked_pd): ''' The purpose of this function is to count the labeled pairs in the marked folder. ''' n_total = len (np.unique(marked_pd[ 'z_cluster' ])) n_positive = len (np.unique(marked_pd[marked_pd[ 'z_isMatch' ]== 1 ][ 'z_cluster' ])) n_negative = len (np.unique(marked_pd[marked_pd[ 'z_isMatch' ]== 0 ][ 'z_cluster' ])) return n_positive, n_negative, n_total # setup widget available_labels = { 'No Match' : 0 , 'Match' : 1 , 'Uncertain' : 2 } # dbutils.widgets.dropdown('label', 'Uncertain', available_labels.keys(), 'Is this pair a match?') 3. Zinggã®èšå®ïŒConfigïŒãå®çŸ© Zinggã§ã¯ã¢ãã«ãäœæããéãå¿
èŠãªèšå®æ
å ±ã¯ãã¹ãŠ Config ïŒå®äœã¯ JSON ãªããžã§ã¯ãïŒãä»ããŠæž¡ãããšã«ãªããŸãã https://docs.zingg.ai/latest/stepbystep/configuration #Zinggãåäœãããã®å¿
èŠãªConfigã®äœæããµããŒãããã¯ã©ã¹ args = Arguments() #ãã¢ãã«åãZinggã®äœæ¥ã¯ãŒã¯ã¹ããŒã¹ãã»ãã args.setModelId(modelId) args.setZinggDir(zinggDir) 4. Zinggã®ã€ã³ããããã¢ãŠããããå
ãèšå®ãã ã€ã³ãããïŒ åå¯ã 察象ã㌠ã¿ã»ãã ã®æ ŒçŽå
ïŒãšã¢ãŠããããïŒåŠççµæã®åºåå
ïŒãèšå®ããŸã # åå¯ã察象ã®ããŒã¿ã»ãã # åè¿°ããŸãããå€éšããŒãã«ãšããŠç»é²ãããŠããå¿
èŠããããŸãã input_path = spark.sql( "DESCRIBE DETAIL catalog.schema.table" ).select( 'location' ).collect()[ 0 ][ 'location' ] # Pipeãšããã¯ã©ã¹ãããã€ã³ãããã®ããŒã¿ã»ãããæå®ãã inputPipe = Pipe(name= 'input' , format = 'delta' ) inputPipe.addProperty( 'path' , input_path ) # Argumentsã«è¿œå args.setData(inputPipe) # ã¢ãŠãããããšãªãããŒã¿ã»ãããã¢ãã«ãæ ŒçŽãããDBFSäžã®ãã¹ output_path = zinggDir + "/" + modelId + "/outputData" # Pipeãšããã¯ã©ã¹ãããã¢ãŠããããã®ããŒã¿ã»ãããæå®ãã outputPipe = Pipe( "resultOutput" , format = "delta" ) outputPipe.addProperty( "path" , output_path) # Argumentsã«è¿œå args.setOutput(outputPipe) 5. ãã£ãŒã«ãã®å±æ§ãæå®ãã æ¬¡ã«ãZinggããããã³ã°åŠçã®äžã§ãã£ãŒã«ãã®åãæ±ãã«ã€ããŠå€æããããã«ã䜿çšãããã£ãŒã«ãæ¯ã«å±æ§ãæå®ããŸãã ãããã®ã¿ã€ãã¯ä»¥äžã®äžããæå®ã§ããŸãã詳现ã¯Zinggã®ããã¥ã¡ã³ããåç
§ããŠãã ããã https://docs.zingg.ai/latest/stepbystep/configuration/field-definitions ãããã¿ã€ã 説æ é©çšå¯Ÿè±¡ FUZZY ã¿ã€ããç¥èªããã®ä»ã®ããªãšãŒã·ã§ã³ãå«ã倧ãŸããªãããã string, integer, long, double, date EXACT ããªãšãŒã·ã§ã³ã蚱容ããªããåœã³ãŒããPINã³ãŒããããã³ããªãšãŒã·ã§ã³ãæ³å®ãããªããã®ä»ã®ã«ããŽãªå€æ°ã«é©ããŠããŸãã string, integer, long, date, boolean DONT_USE åºåã«ã¯è¡šç€ºãããŸãããèšç®ã¯è¡ãããŸãããåºåã«å¿
èŠãªIDãªã©ã®ãã£ãŒã«ãã«åœ¹ç«ã¡ãŸãã showConcise ãtrueã«èšå®ãããŠããå ŽåãDONT_USEãã£ãŒã«ãã¯ã©ãã«ä»ãäžã«ãŠãŒã¶ãŒã«è¡šç€ºãããŸããã any EMAIL @æåã®åã®ã¡ãŒã«ã®IDéšåã®ã¿ããããããŸãã any PINCODE xxxxx-xxxxã®ãããªPINã³ãŒãããããããŸãïŒxxxxxä»ãïŒã string NULL_OR_BLANK ããã©ã«ãã§ã¯ãZinggã¯nullãããããšããŠæ±ããŸãããFUZZYã®ãããªå¥ã®ãããã¿ã€ããæã€ãã£ãŒã«ãã«ããïŒNULL_OR_BLANKïŒã远å ãããšãZinggã¯nullå€ã®æ©èœãæ§ç¯ããŠåŠç¿ããŸãã string, integer, long, date, boolean TEXT 2ã€ã®æååéã®åèªã®éè€ãæ¯èŒããŸãã ã¿ã€ããã¹ ãããŸããªãèšè¿°çãªãã£ãŒã«ãã«é©ããŠããŸãã string NUMERIC æååããæ°åãæœåºããäž¡æ¹ã®æååéã§åããã®ãããã€ããããæ¯èŒããŸãïŒäŸïŒã¢ããŒãçªå·ïŒã string NUMERIC_WITH_UNITS 補åã³ãŒããŸãã¯åäœä»ãã®æ°åãæœåºããŸãïŒäŸïŒ16gbã®æååïŒãäž¡æ¹ã®æååéã§åããã®ãããã€ããããæ¯èŒããŸãã string ONLY_ ALPHABETS _EXACT ã¢ã«ãã¡ãããæåã®ã¿ãèŠãŠãããããå®å
šã«åãã§ãããã©ãããæ¯èŒããŸããæååå
ã®æ°åãåé¡ã§ã¯ãªãå ŽåïŒäŸïŒå»ºç©ã調ã¹ãŠãããããã©ããçªå·ãç¡èŠãããå ŽåïŒã string ONLY_ ALPHABETS _FUZZY æååå
ã®æ°åãç¡èŠãããããŸããªæ¯èŒãå®è¡ããŸãã ã¿ã€ããã¹ ã®ããäœæãªã©ã®ãã£ãŒã«ãã«åœ¹ç«ã¡ãŸããNUMERICã䜿çšããŠçªå°ãåå¥ã«èª¿ã¹ããå Žåã string MAPPING_(FILENAME) ã¿ã€ããã¹ ãç¥èªãããã³æå®ãããå€ã«å¯Ÿãããã®ä»ã®ããªãšãŒã·ã§ã³ãšã®ãããã³ã°ãããã¯ããŒã ãæ§å¥ãããã³äŒç€Ÿç¥èªãªã©ã®å Žåã«äœ¿çšã§ããŸããããšãã°ã IBM ãInt. Biz. MachineãInternational Business Machineã«ãMã0ãMrãMaleã«ïŒã€ãŸããè€æ°ã®ãšã³ãã£ãã£ãåäžã®ãšã³ãã£ãã£ã«ïŒ ãããã³ã° ããå Žåã string ããŒãããã¯ã§ã¯æ¬¡ã®ããã«èšå®ããŸãã # ãã£ãŒã«ãæ¯ã«ãããã¿ã€ããæå®ããŸãã # åå¯ãåŠçããããããŒã¿ã»ããæ¯ã«å®éã¯èšå®ããŠãã ãã rec_id = FieldDefinition( "rec_id" , "string" , MatchType.DONT_USE) name = FieldDefinition( "name" , "string" , MatchType.FUZZY) ## Match Typeã¯è€æ°æå®å¯èœã§ãã email = FieldDefinition( "email" , "string" , [MatchType.FUZZY,MatchType.EMAIL]) dob = FieldDefinition( "dob" , "string" , [MatchType.FUZZY,MatchType.NUMERIC]) address = FieldDefinition( "address" , "string" , MatchType.FUZZY) phone = FieldDefinition( "phone" , "string" , [MatchType.FUZZY,MatchType.NUMERIC]) # ãã£ãŒã«ãã®å®çŸ©ãArrayã«ãã fieldDefs = [rec_id, name, email, dob, address, phone] # Argumentsã«èšå® args.setFieldDefinition(fieldDefs) 6. ãã©ã¡ãŒã¿ãèšå®ãã æ¬¡ã«ãSparkäžã§åæ£åŠçãè¡ãéã®ãã©ã¡ãŒã¿ãèšå®ããŸããå
·äœçã«ã¯ numPartitions ãš labelDataSampleSize ãèšå®ããŸãã numPartitions numPartitions ã¯ãããŒã¿ã ã¯ã©ã¹ã¿ éã§ã©ã®ããã«åå²ããããå®çŸ©ããŸãã Zinggããã¥ã¡ã³ã ãåç
§ããŠãããŒã¿ãš ã¯ã©ã¹ã¿ ãµã€ãºã«å¿ããŠå€æŽããŠãã ããã labelDataSampleSize labelDataSampleSize ã§ã¯ãåŸè¿°ãã findTrainingData åŠçã§ãµã³ãã«ããå²åãæå®ã§ããŸãããµã³ããªã³ã°ããå²åã倧ããã»ãããã©ããªã³ã°ããéã«ã±ãŒã¹ãšããŠçžå¿ãããã¢ãåºåã§ããå¯èœæ§ãé«ããªããŸãããåŠçã®æéãé·ããªããããå Žåãããã®ã§ã1/10ã«æžãããªã©èª¿æŽãè¡ã£ãŠãã ããã # Sparkã®ããŒãã£ã·ã§ã³æ°ãããã¥ã¡ã³ãããæšå¥šã¯ã³ã¢æ° à 20~30ãšã®ããš num_partitions = sc.defaultParallelism * 20 args.setNumPartitions( num_partitions ) # ã¢ãã«ã®åŠç¿ã«äœ¿çšããããŒã¿ã®å²åã # ãµã³ãã«ãµã€ãºãå°ããä¿ã¡ãååãªãšããžã±ãŒã¹ãè¿
éã«æ€åºã§ããããã0.0001ïœ0.1ã®éã§èª¿æŽããŠãã ããã # ãµã³ãã«ãµã€ãºã倧ãããšããžã§ããfindTrainingDataãµã³ãã«ã粟æ»ããã®ã«æéãããããŸãããµã€ãºãå°ãããããšãZinggãé©åãªãšããžã±ãŒã¹ãèŠã€ããããªãå¯èœæ§ããããŸãã args.setLabelDataSampleSize( 0.05 ) 7. ã ã¬ãŒãã³ ã°ããŒã¿äœæãžã§ããå®è¡ãã ãããŸã§ã®èšå®ãçšããŠãã ã¬ãŒãã³ ã°ããŒã¿ïŒã©ããªã³ã°åè£ãã¢ïŒãäœæãããžã§ãïŒ findTrainingData ãã§ãŒãºïŒãå®è¡ããŸãã options = ClientOptions([ClientOptions.PHASE,"findTrainingData"]) # ãã¬ãŒãã³ã°ããŒã¿ãç¹å®ãããžã§ããå®è¡ zingg = ZinggWithSpark(args, options) zingg.initAndExecute() 8. ã©ããªã³ã°ã«äœ¿çšããã㌠ã¿ã»ãã ãäœæãã ã ã¬ãŒãã³ ã°ããŒã¿ã®äœæãçµãã£ãããã©ããªã³ã°ã«äœ¿çšããã㌠ã¿ã»ãã ãäœæããŸãã ã¡ãªã¿ã«ãã¹ããã7ãš8ã®åŠçïŒã ã¬ãŒãã³ ã°ããŒã¿äœæãšã©ããªã³ã°çšã㌠ã¿ã»ãã æºåïŒã¯ã findAndLabel ãšããåäžã®ãã§ãŒãºæå®ã§é£ç¶å®è¡ãå¯èœã§ããããããã ã¬ãŒãã³ ã°ããŒã¿ã®äœæã«ã¯æéããããå Žåããããããããã§ã¯éäžããåéããããããã«ã¹ããããåããŠèšè¿°ããŠããŸãã options = ClientOptions([ClientOptions.PHASE, "label" ]) # ã©ããªã³ã°ã«äœ¿çšããããŒã¿ã»ãããäœæãããžã§ãã宿œ zingg = ZinggWithSpark(args, options) zingg.init() 次ã«ãã©ããªã³ã°ã®åè£ãšãªããã¢ãäœä»¶ããã確èªããŸãã # get candidate pairs candidate_pairs_pd = getPandasDfFromDs(zingg.getUnmarkedRecords()) # if no candidate pairs, run job and wait if candidate_pairs_pd.shape[ 0 ] == 0 : print ( 'No unlabeled candidate pairs found. Run findTraining job ...' ) else : # get list of pairs (as identified by z_cluster) to label z_clusters = list (np.unique(candidate_pairs_pd[ 'z_cluster' ])) # identify last reviewed cluster last_z_cluster = '' # none yet # print candidate pair stats print ( '{0} candidate pairs found for labeling' .format( len (z_clusters))) ã©ããªã³ã°ã®åè£ãšããŠããã€ãã¢ãããåºåãããŸããïŒç»åã§ã¯ã©ããªã³ã°ã®åè£ãšããŠ20ãã¢ããããšãããããŸãïŒ 9. ã©ããªã³ã°ãè¡ã ãããããæœåºãããåè£ãã¢ã«å¯ŸããŠå®éã«ã©ããªã³ã°ãè¡ããŸãã ããŒãããã¯äžã«åè£ãã¢ãHTML圢åŒã§è¡šç€ºãããåãã¢ã«å¯ŸããŠãUncertainïŒäžæïŒããMatchïŒäžèŽïŒããNo MatchïŒäžäžèŽïŒãã®ããããããã¿ã³ãæŒãããšã§ã©ããªã³ã°äœæ¥ãè¡ããŸãã ïŒDatabricksãããŒãããã¯æ©èœãæäŸããŠãã匷ã¿ã§ããïŒïŒ # Label Training Set # define variable to avoid duplicate saves ready_for_save = False # user-friendly labels and corresponding zingg numerical value # (the order in the dictionary affects how displayed below) LABELS = { 'Uncertain' : 2 , 'Match' : 1 , 'No Match' : 0 } # GET CANDIDATE PAIRS # ======================================================== #candidate_pairs_pd = get_candidate_pairs() n_pairs = int (candidate_pairs_pd.shape[ 0 ]/ 2 ) # ======================================================== # DEFINE IPYWIDGET DISPLAY # ======================================================== display_pd = candidate_pairs_pd.drop( labels=[ 'z_zid' , 'z_prediction' , 'z_score' , 'z_isMatch' , 'z_zsource' ], axis= 1 ) # define header to be used with each displayed pair html_prefix = "<p><span style='font-family:Courier New,Courier,monospace'>" html_suffix = "</p></span>" header = widgets.HTML(value=f "{html_prefix}<b>" + "<br />" .join([ str (i)+ "&nbsp;&nbsp;" for i in display_pd.columns.to_list()]) + f "</b>{html_suffix}" ) # initialize display vContainers = [] vContainers.append(widgets.HTML(value=f '<h2>Indicate if each of the {n_pairs} record pairs is a match or not</h2></p>' )) # for each set of pairs for n in range (n_pairs): # get candidate records candidate_left = display_pd.loc[ 2 *n].to_list() candidate_right = display_pd.loc[( 2 *n)+ 1 ].to_list() # define grid to hold values html = '' for i in range (display_pd.shape[ 1 ]): # get column name column_name = display_pd.columns[i] # if field is image if column_name == 'image_path' : # define row header html += '<tr>' html += '<td><b>image</b></td>' # read left image to encoded string l_endcode = '' if candidate_left[i] != '' : with open (candidate_left[i], "rb" ) as l_file: l_encode = base64.b64encode( l_file.read() ).decode() # read right image to encoded string r_encode = '' if candidate_right[i] != '' : with open (candidate_right[i], "rb" ) as r_file: r_encode = base64.b64encode( r_file.read() ).decode() # present images html += f '<td><img src="data:image/png;base64,{l_encode}"></td>' html += f '<td><img src="data:image/png;base64,{r_encode}"></td>' html += '</tr>' elif column_name != 'image_path' : # display text values if column_name == 'z_cluster' : z_cluster = candidate_left[i] html += '<tr>' html += f '<td style="width:10%"><b>{column_name}</b></td>' html += f '<td style="width:45%">{str(candidate_left[i])}</td>' html += f '<td style="width:45%">{str(candidate_right[i])}</td>' html += '</tr>' # insert data table table = widgets.HTML(value=f '<table data-title="{z_cluster}" style="width:100%;border-collapse:collapse" border="1">' +html+ '</table>' ) z_cluster = None # assign label options to pair label = widgets.ToggleButtons( options=LABELS.keys(), button_style= 'info' ) # define blank line between displayed pair and next blankLine=widgets.HTML(value= '<br>' ) # append pair, label and blank line to widget structure vContainers.append(widgets.VBox(children=[table, label, blankLine])) # present widget display(widgets.VBox(children=vContainers)) # ======================================================== # mark flag to allow save ready_for_save = True ããŒãããã¯ã®ã¢ãŠããããã§ã¯æ¬¡ã®ãããªHTMLã衚瀺ãããŸãã å®éã«ã¬ã³ãŒããäžèŽããŠããããäžèŽããŠããªããã人ã®ç®ã§1ã€ã²ãšã€ãã§ãã¯ãè¡ãããã¿ã³ãæŒããŠã©ããªã³ã°ãè¡ã£ãŠãããŸãã ã¹ããã8ã§äœæããããã¢åã®ãã§ãã¯ãçµäºããããæ¬¡ã®ã»ã«ãå®è¡ããå
容ãä¿åããŸãã if not ready_for_save: print ( 'No labels have been assigned. Run the previous cell to create candidate pairs and assign labels to them before re-running this cell.' ) else : # ASSIGN LABEL VALUE TO CANDIDATE PAIRS IN DATAFRAME # ======================================================== # for each pair in displayed widget for pair in vContainers[ 1 :]: # get pair and assigned label html_content = pair.children[ 1 ].get_interact_value() # the displayed pair as html user_assigned_label = pair.children[ 1 ].get_interact_value() # the assigned label # extract candidate pair id from html pair content start = pair.children[ 0 ].value.find( 'data-title="' ) if start > 0 : start += len ( 'data-title="' ) end = pair.children[ 0 ].value.find( '"' , start+ 2 ) pair_id = pair.children[ 0 ].value[start:end] # assign label to candidate pair entry in dataframe candidate_pairs_pd.loc[candidate_pairs_pd[ 'z_cluster' ]==pair_id, 'z_isMatch' ] = LABELS.get(user_assigned_label) # ======================================================== # SAVE LABELED DATA TO ZINGG FOLDER # ======================================================== # make target directory if needed dbutils.fs.mkdirs(MARKED_DIR) # save label assignments # save labels zingg.writeLabelledOutputFromPandas(candidate_pairs_pd,args) # count labels accumulated marked_pd_df = getPandasDfFromDs(zingg.getMarkedRecords()) n_pos, n_neg, n_tot = count_labeled_pairs(marked_pd_df) print (f 'You have accumulated {n_pos} pairs labeled as positive matches.' ) print (f 'You have accumulated {n_neg} pairs labeled as not matches.' ) print ( "If you need more pairs to label, re-run the cell for 'findTrainingData'" ) # ======================================================== # save completed ready_for_save = False äžèšã®ã»ã«ãå®è¡ãããšããããŸã§ã«ã©ããªã³ã°ãããã¢ã®ç·æ°ïŒäžèŽãäžäžèŽããšã®ä»¶æ°ïŒãåºåããã鲿ã確èªã§ããŸãã Zinggã®ããã¥ã¡ã³ã ã§ã¯ãæäœã§ãããããšå€æã§ãããã¢ã30~40å¿
èŠãšãããããããšå€æããæ°ãè¶³ããªãå Žåã¯ããããããŠã7. ã ã¬ãŒãã³ ã°ããŒã¿äœæãžã§ãå®è¡ãããããç¹°ãè¿ããååãªæåž«ããŒã¿ãåéããŸãã ãŸããåè£ãšãªããã¢ã®è³ªãäœãå ŽåïŒæããã«å¥ã®äººããªã¹ãã¢ãããããŠããå Žåãªã©ïŒã labelDataSampleSize ã®å²åãå°ããå¯èœæ§ãããã®ã§ãå¿
èŠã«å¿ããŠãã¡ããæ€èšããŠã¿ãŠãã ããã 10. ã¢ãã«ã®ã ã¬ãŒãã³ ã° ååãªæ°ã®ã©ããªã³ã°æžã¿æåž«ããŒã¿ãåéã§ããããããããåºã«ãããã³ã°ã¢ãã«ãåŠç¿ããŸãïŒ trainMatch ãã§ãŒãºïŒ options = ClientOptions([ClientOptions.PHASE, "trainMatch" ]) #ãã¬ãŒãã³ã°ãžã§ãã宿œããŸãïŒæéããããŸãïŒ zingg = ZinggWithSpark(args, options) zingg.initAndExecute() 11. ãããã³ã°çµæã確èªãã ã¢ãã«åŠç¿ãå®äºãããããã®ã¢ãã«ãçšã㊠åå¯ã 察象ã®å
šããŒã¿ã«å¯ŸããŠäºæž¬ãå®è¡ããã©ã®ã¬ã³ãŒããã¢ãåäžãšã³ãã£ãã£ãšããŠå€æããããã確èªããŸãã äºæž¬çµæãåºåãããã㌠ã¿ã»ãã ã¯ãåè¿°ã®Zinggèšå®ïŒ outputPipe ïŒã§æå®ãããã¹ã«æ ŒçŽãããŸãã ãã®åºåã㌠ã¿ã»ãã ã«ã¯ãå
¥åããŒã¿ã«å ããŠãZinggãä»äžãã ã¯ã©ã¹ã¿ IDãé¡äŒŒåºŠã¹ã³ã¢ãªã©ã®æ
å ±ãå«ãŸããŸãã from pyspark.sql.types import StructType, StructField, IntegerType, FloatType, StringType # ããŒãã«ã®ååŸ outputDF = spark.read.load(zinggDir + "/" + modelId + "/outputData/" ) # å
容ã®ç¢ºèª display(outputDF.orderBy( "z_cluster" ).head( 25 )) å
·äœçã«ã¯ã以äžã®åã远å ãããŠããŸãã z_minScore ãã®ã¬ã³ãŒããåã ã¯ã©ã¹ã¿ å
ã®å¥ã®ã¬ã³ãŒããšæã€æå°ã®é¡äŒŒæ§ã¹ã³ã¢ã衚ããŸãã z_maxScore ãã®ã¬ã³ãŒããåã ã¯ã©ã¹ã¿ å
ã®å¥ã®ã¬ã³ãŒããšæã€æå€§ã®é¡äŒŒæ§ã¹ã³ã¢ã衚ããŸãã z_cluster Zinggã«ãã£ãŠå²ãåœãŠããã ã¯ã©ã¹ã¿ ãŒã®IDãåäžãšã³ãã£ãã£ãšå€å®ãããã¬ã³ãŒã矀ã«ã¯ãå
±éã®z_cluster IDãä»äžãããŸãã z_cluster ããã®ãŸãŸäžæã®ãšã³ãã£ãã£ã衚ããŠããŒã¯IDãšããŠäœ¿çšã§ããŸããã z_minScore ã z_maxScore ãå
ã« éŸå€ ããŠãŒã¶åŽã§å®çŸ©ïŒXX%以äžã¯äžèŽãšãããXX ~ YY%ã¯ãŠãŒã¶ã§æ¹ããŠç®æ€ãYY%以äžã¯å¥äººãšããçã
...ïŒããä»»æã®ãªãã¬ãŒã·ã§ã³ãããããšãã§ããŸãã ïŒè©³çްã«ã€ããŠã¯ã以äžã®Zinggå
¬åŒããã¥ã¡ã³ãããåç
§ãã ããïŒ https://www.zingg.ai/documentation-article/step-by-step-identity-resolution-with-python-and-zingg 以äžã§ åå¯ã ã®äžé£ã®ã¹ããããå
šãŠçµäºããŸããïŒ ãŸãšã æ¬ããã°èšäºã§ã¯ãZinggã©ã€ãã©ãªãçšãã確ççãããã³ã°ã®å
·äœçãªææ³ã玹ä»ããŸããã ä»ååãäžããZinggã®ãããªæåž«ãã æ©æ¢°åŠç¿ ãæŽ»çšãã åå¯ã ã©ã€ãã©ãªããDatabricksã®ããŒãããã¯æ©èœãããããŒã¿æºåããã©ããªã³ã°ãã¢ãã«åŠç¿ãæšè«ãŸã§ãäžè²«ããã¯ãŒã¯ããŒããšããŠå¹ççã«å®è¡ã§ããŸãã ãŸããä»åã®èšäºã§ã¯1ã€ã®ã㌠ã¿ã»ãã å
ã§ åå¯ã ããæ¹æ³ã«ã€ããŠåãäžããŸããããä»ã«ãZinggã䜿çšããŠãå¢åããã㌠ã¿ã»ãã ã«å¯ŸããŠç¶ç¶çã« åå¯ã åŠçã宿œã§ããã ïŒ runIncremental ã ãšã³ã¿ãŒãã©ã€ãº çã§å©çšå¯èœïŒãè€æ°ã®ã㌠ã¿ã»ãã éã§äžèŽãåŸãã¬ã³ãŒããæœåºãå®è¡å¯èœã§ãïŒ link ïŒã®ã§ãããèå³ãããæ¹ã¯æ¯éå
¬åŒããã¥ã¡ã³ãã®æ¹ãã芧ãã ããã äžèšäºã«ããã åå¯ã ã®æŠèŠããDatabricksäžã§Splinkã«ãã æ±ºå®è« çææ³ãZinggã«ãã確ççææ³ãŸã§ åå¯ã ã®ã¢ãããŒãã«ã€ããŠç޹ä»ããŸãããããããã ã£ãã§ããããïŒ ä»åã®äžé£ã®èšäºããçæ§ã®ããŒã¿æŽ»çšã åå¯ã æ¥åã®äžå©ãšãªãã°å¹žãã§ãã ç§ãã¡ã¯äžç·ã«åããŠããã仲éãåéããŠããŸãïŒ é»éç·ç ãã£ãªã¢æ¡çšãµã€ã é»éç·ç æ°åæ¡çšãµã€ã å·çïŒ @kumakura.koki ãã¬ãã¥ãŒïŒ @kinjo.ryuki ïŒ Shodo ã§å·çãããŸãã ïŒ