-
Notifications
You must be signed in to change notification settings - Fork 0
Prototype Bulkloader
This document provides some documentation for a very early version of the engine.py bulkloader prototype.
First you'll need to clone the project to your machine:
git clone git@github.com:VertNet/Darwin-Core-Engine.git
The script is called engine.py and is located in the Darwin-Core-Engine directory. Type ./engine.py to view the command options:
Usage: engine.py [options] <action>
Action must be one of:
bulkload: Bulkloads report.csv to Google App Engine datastore.
deltas: Calculate deltas for a file.
help: Print help for a specific action.
report: Creates a report.
Use 'help <action>' for a detailed description.
Options:
-h, --help Show the help message and exit.
-v, --verbose Print info level logs.
--config_file=FILE Bulkload YAML config file.
--filename=FILE CSV file with data to bulkload.
--url=URL URL endpoint to /remote_api to bulkload to.
--num_threads=NUM_THREADS
Number of threads to transfer records with.
-l, --localhost Shortcut for bulkloading to
http://localhost:8080/_ah/remote_api
-b SIZE, --batch_size=SIZE
Batch size for processing.
-f FILE, --csv_file=FILE
Input CSV file.
-p PUBLISHER, --publisher_name=PUBLISHER
VertNet publisher name.
-c COLLECTION, --collection_name=COLLECTION
VertNet publisher collection name.
-s SOURCEID, --source_id=SOURCEID
Column name that contains the source record id.
The script lets you calculate deltas for a CSV file. Huh? Basically, you run the script with multiple CSV files and it will figure out what records are new, updated, and deleted. Type ./engine.py help deltas for info:
Usage: engine.py [options] deltas <file>
Calculate deltas for a file.
Specify a CSV file, and vn.py will generate deltas for new, updated
and deleted records. Deltas will be persisted via sqlite3.
Options:
-h, --help Show the help message and exit.
-v, --verbose Print info level logs.
-b SIZE, --batch_size=SIZE
Batch size for processing.
-f FILE, --csv_file=FILE
Input CSV file.
-p PUBLISHER, --publisher_name=PUBLISHER
VertNet publisher name.
-c COLLECTION, --collection_name=COLLECTION
VertNet publisher collection name.
-s SOURCEID, --source_id=SOURCEID
Column name that contains the source record id.
Let's calculate some deltas! Create a file named data.csv with the following content:
occurrenceid,genus,year,country
H-1,bufo,1988,china
F-1,salmo,1900,united states
M-1,tuco,1985,argentina
B-1,vultur,2010,chile
H-2,atrox,1985,mexico
Now let's run the command. Note that provider, collection, and source_id are required command-line options. For this example the provider and collection can be any names you want, but the source_id needs to be "occurrence_id". Here we go. The following command says "calculate deltas" for publisher "My Institution" and collection "Vert Collection" where the source_id is the field "occurrenceid" in the file "data.csv":
./engine.py deltas -p 'My Institution' -c 'Vert Collection' -s occurrenceid -f data.csv
You should see this output:
Calculating deltas for data.csv
Processing incoming records
5...
Processed 5 records
Checking for new records
5...
5 new records found
Checking for updated records
No updated records found
Checking for deleted records
No deleted records found
Boom! What just happened? The script saved the records in a table named cache in a sqlite3 database named bulk.sqlite3.db in the current directory . Check it! (Hint: You can exit sqlite with the command .quit.)
$ sqlite3 bulk.sqlite3.db
sqlite> select * from cache;
agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2gtMQw|229c8fa0b67a3ceac4458becedd9757e46a36fdbb30aebff2bd02f06|{"country": "china", "genus": "bufo", "occurrenceid": "H-1", "year": "1988"}|new
agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2YtMQw|6e468c5de2e71d24f2082faf4c96c9dc02afaf2f35eeee12a9acbb62|{"country": "united states", "genus": "salmo", "occurrenceid": "F-1", "year": "1900"}|new
agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA20tMQw|e26004ac1621859df780df85aafdbfb5b0a4dbf130a29f68159eb2d8|{"country": "argentina", "genus": "tuco", "occurrenceid": "M-1", "year": "1985"}|new
agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2ItMQw|a1ea1df324d2a38a7a576db5024a0c3f12590f7273178cda16ec9c8d|{"country": "chile", "genus": "vultur", "occurrenceid": "B-1", "year": "2010"}|new
agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2gtMgw|36cc22d60c1a910b042a4005d690a104546fba1fa3366430a21d9c5a|{"country": "m\u00e9xico", "genus": "atrox", "occurrenceid": "H-2", "year": "1985"}|new
Here's what the cache table looks like (hint: type .schema cache while in sqlite3):
| Column | Description | Example |
| reckey | The unique record key for use in App Engine datastore | agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA20tMQw |
| rechash | sha digest of record contents | e26004ac1621859df780df85aafdbfb5b0a4dbf130a29f68159eb2d8 |
| recjson | the record as a JSON string | {"country": "argentina", "genus": "tuco", "occurrenceid": "M-1", "year": "1985"} |
| recstate | the record state (new, updated, deleted) | new |
So when a CSV file is processed, all records are first stored in a temporary table with the same schema as the cache table. Then, to figure out the deltas, we do 3 simple joins and update cache.recstate based on what we find:
# New records:
SELECT * FROM tmp LEFT OUTER JOIN cache USING (reckey) WHERE cache.reckey is null
# Updated records:
SELECT * FROM tmp, cache WHERE tmp.reckey = cache.reckey AND tmp.rechash <> cache.rechash
# Deleted records:
SELECT * FROM cache LEFT OUTER JOIN tmp USING (reckey) WHERE tmp.reckey is null
This all cool, but I want to see a CSV file with the deltas! Enter the report command, which generates a CSV report file called report.csv:
./engine.py report
You will see the following output:
Creating a report
Creating report
Report saved to report.csv
Open report.csv and you will see all records that have are new, updated, or deleted. In this case, there are only new records:
recstate,reckey,rechash,recjson
new,agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2gtMQw,229c8fa0b67a3ceac4458becedd9757e46a36fdbb30aebff2bd02f06,"{""country"": ""china"", ""genus"": ""bufo"", ""occurrenceid"": ""H-1"", ""year"": ""1988""}"
new,agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2YtMQw,6e468c5de2e71d24f2082faf4c96c9dc02afaf2f35eeee12a9acbb62,"{""country"": ""united states"", ""genus"": ""salmo"", ""occurrenceid"": ""F-1"", ""year"": ""1900""}"
new,agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA20tMQw,e26004ac1621859df780df85aafdbfb5b0a4dbf130a29f68159eb2d8,"{""country"": ""argentina"", ""genus"": ""tuco"", ""occurrenceid"": ""M-1"", ""year"": ""1985""}"
new,agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2ItMQw,a1ea1df324d2a38a7a576db5024a0c3f12590f7273178cda16ec9c8d,"{""country"": ""chile"", ""genus"": ""vultur"", ""occurrenceid"": ""B-1"", ""year"": ""2010""}"
new,agFfcksLEglQdWJsaXNoZXIiDk15IEluc3RpdHV0aW9uDAsSCkNvbGxlY3Rpb24iD1ZlcnQgQ29sbGVjdGlvbgwLEgZSZWNvcmQiA2gtMgw,36cc22d60c1a910b042a4005d690a104546fba1fa3366430a21d9c5a,"{""country"": ""m\u00e9xico"", ""genus"": ""atrox"", ""occurrenceid"": ""H-2"", ""year"": ""1985""}"
If you run ./engine.py with exactly the same command-line arguments as before without having changed data.csv:
./engine.py deltas -p 'My Institution' -c 'Vert Collection' -s occurrenceid -f data.csv
you'll notice that the output shows no changes.
Calculating deltas for data.csv
Processing incoming records
5...
Processed 5 records
Checking for new records
No new records found
Checking for updated records
No updated records found
Checking for deleted records
No deleted records found
Next run ./engine.py deltas after making modifications to the data.csv file or by calling ./engine.py pointing to a different CSV file. Run ./engine.py report after each delta to see how the local database is keeping track of the state of changes (see States-Transitions-for-engine.py-deltas).
Ok, so now we have report.csv that contains new, updated, and deleted records, right? Lets bulkload it to the App Engine datastore so that we can start querying it. Type ./engine.py help bulkload for info:
Usage: engine.py [options] bulkload <file>
Bulkloads report.csv to Google App Engine datastore.
Specify a CSV file and a URL, and vn.py will bulkload it to the
Google App Engine datastore.
Options:
-h, --help Show the help message and exit.
-v, --verbose Print info level logs.
--config_file=FILE Bulkload YAML config file.
--filename=FILE CSV file with data to bulkload.
--url=URL URL endpoint to /remote_api to bulkload to.
--num_threads=NUM_THREADS
Number of threads to transfer records with.
--batch_size=BATCH_SIZE
Number of records to pst in each request.
-l, --localhost Shortcut for bulkloading to
http://localhost:8080/_ah/remote_api
Let's bulkload report.csv locally to the App Engine development server. First make sure you have the full path to /full/system/path/to/VertNet/Darwin-Core-Engine/lib/google_appengine on your environment $PATH. Then to start the development server via terminal:
- Go into the
/VertNet/Darwin-Core-Engine/appdirectory - Type the
dev_appserver.py -c .command
If the development server started correctly, you should see output such as this:
INFO 2011-08-05 21:26:14,594 appengine_rpc.py:159] Server: appengine.google.com
INFO 2011-08-05 21:26:14,596 appcfg.py:456] Checking for updates to the SDK.
INFO 2011-08-05 21:26:22,962 rdbms_sqlite.py:58] Connecting to SQLite database
INFO 2011-08-05 21:26:22,980 dev_appserver_multiprocess.py:637] Running application vert-net on port 8080: http://localhost:8080
Ok, our dev server is running, so now in a new terminal window go into the /VertNet/Darwin-Core-Engine directory and type the following command:
./engine.py bulkload --filename=report.csv --config_file=darwin_core_engine/publishing/tools/bulkload.yaml --url=http://localhost:8080/_ah/remote_api
You will be prompted to enter an email address, but you can leave this blank and just hit the enter key. After the script completes, the contents of report.csv have been bulkloaded to the development server. You can verify this by accessing the datastore viewer in the admin console at http://localhost/_ah/admin.
Now that you have data in your development server, you can query against it using the prototype search API at your http://localhost/api/search endpoint!