############################################################################ # # # Imports & Constants # # # ############################################################################ import argparse import gpudb import sys import json from tabulate import tabulate CSV_FILE = "nyc_neighborhood.csv" OPTION_NO_DROP_ERROR = {"no_error_if_not_exists": "true"} OPTION_NO_CREATE_ERROR = {"no_error_if_exists": "true"} SCHEMA = "graph_s_nyctaxi_multi_route" TABLE_NYC_N = SCHEMA + ".nyc_neighborhood" TABLE_TAXI = "demo.nyctaxi" TABLE_TAXI_EW = SCHEMA + ".nyctaxi_edges_weights_wkt" JOIN_TAXI = SCHEMA + ".taxi_tables_joined" GRAPH_T = SCHEMA + ".nyctaxi_graph_wkt" TABLE_GRAPH_T_MRSOLVED = GRAPH_T + "_multiple_routing_solved" ############################################################################ # # # Table Setup Function # # # ############################################################################ def table_setup(data_dir): """ Setup necessary source tables for graph solver examples. """ print("\n===========") print("TABLE SETUP") print("===========\n") # Clear any related tables in case they already exist kinetica.clear_table(table_name=JOIN_TAXI, options=OPTION_NO_DROP_ERROR) kinetica.clear_table(table_name=TABLE_NYC_N, options=OPTION_NO_DROP_ERROR) kinetica.clear_table(table_name=TABLE_TAXI_EW, options=OPTION_NO_DROP_ERROR) # Create the graph example schema, if it doesn't exist kinetica.create_schema(SCHEMA, options=OPTION_NO_CREATE_ERROR) # Create the NYC neighborhood table try: table_nycn_obj = gpudb.GPUdbTable( _type = [ ["gid", "int"], ["geom", "string", "wkt"], ["CTLabel", "string", "char16"], ["BoroCode", "string", "char16"], ["BoroName", "string", "char16"], ["CT2010", "string", "char16"], ["BoroCT2010", "string", "char16"], ["CDEligibil", "string", "char16"], ["NTACode", "string", "char16"], ["NTAName", "string", "char64"], ["PUMA", "string", "char16"], ["Shape_Leng", "double"], ["Shape_Area", "double"] ], name = TABLE_NYC_N, db = kinetica, options = {"is_replicated": "true"} ) print("{} table object successfully created.".format(TABLE_NYC_N)) except gpudb.GPUdbException as e: print("{} table object creation failure: {}".format(TABLE_NYC_N, str(e))) # Insert records from a CSV file into the NYC neighborhood table csv_path = data_dir + "/" + CSV_FILE print("Creating records from the " + CSV_FILE + " file.") kinetica.create_directory("data", {"no_error_if_exists":"true"}); kinetica.upload_files("/data/" + CSV_FILE, open(csv_path, "rb").read()) kinetica.insert_records_from_files(TABLE_NYC_N, ["kifs://data/" + CSV_FILE]) print("{} records inserted into {} table.\n".format(table_nycn_obj.size(), TABLE_NYC_N)) # Check to see if 'nyctaxi' table exists print("Checking to see if table {} exists.".format(TABLE_TAXI)) if kinetica.has_table(table_name=TABLE_TAXI)['table_exists']: print("Table {} exists.".format(TABLE_TAXI)) else: print("Table {} does not exist. Please ingest the NYCTaxi data set and try again.".format(TABLE_TAXI)) sys.exit(1) print("") # Join the TABLE_TAXI table to the TABLE_NYC_N table using STXY_CONTAINS # to filter out data that could skew the graph print("Joining {} to {} to filter out data that could skew the taxi graphs.".format(TABLE_TAXI, JOIN_TAXI)) join_taxi_tables_response = kinetica.create_join_table( join_table_name = JOIN_TAXI, table_names = [TABLE_TAXI + " as t", TABLE_NYC_N + " as n"], column_names = [ "CONCAT(CHAR32(pickup_longitude), CHAR32(pickup_latitude)) as pickup_name", "t.pickup_longitude", "t.pickup_latitude", "HASH(t.pickup_longitude + t.pickup_latitude) as pickup_id", "CONCAT(CHAR32(dropoff_longitude), CHAR32(dropoff_latitude)) as dropoff_name", "t.dropoff_longitude", "t.dropoff_latitude", "HASH(t.dropoff_longitude + t.dropoff_latitude) as dropoff_id", "t.total_amount" ], expressions = [ "(STXY_CONTAINS(n.geom, t.pickup_longitude, t.pickup_latitude)) AND" "(STXY_CONTAINS(n.geom, t.dropoff_longitude, t.dropoff_latitude)) " ] )["status_info"]["status"] print("{} view created: {}".format(JOIN_TAXI, join_taxi_tables_response)) print("") # Create a projection to contain the graph edges (based on EDGE_WKTLINE) print("Creating a projection from {} to contain the {} edges.".format(JOIN_TAXI, GRAPH_T)) edges_wkt_response = kinetica.create_projection( table_name = JOIN_TAXI, projection_name = TABLE_TAXI_EW, column_names = [ "REMOVE_NULLABLE(ST_MAKELINE(" "ST_MAKEPOINT(pickup_longitude, pickup_latitude)," "ST_MAKEPOINT(dropoff_longitude, dropoff_latitude)" ")) AS tripwkt", "total_amount" ], options = {} )["status_info"]["status"] print("{} projection created: {}".format(TABLE_TAXI_EW, edges_wkt_response)) # end table_setup() ############################################################################ # # # Graph Setup Function # # # ############################################################################ def graph_setup(): """ Setup graphs to be used in solver examples. """ print("\n===========") print("GRAPH SETUP") print("===========\n") # Create a graph from TABLE_TAXI_EW using WKTLINE print("Creating {}".format(GRAPH_T)) create_t_graph_response = kinetica.create_graph( graph_name = GRAPH_T, directed_graph = False, nodes = [], edges = [ TABLE_TAXI_EW + ".tripwkt AS WKTLINE", TABLE_TAXI_EW + ".total_amount AS WEIGHT_VALUESPECIFIED" ], weights = [], restrictions = [], options = { "recreate": "true", "merge_tolerance": "0.01" } ) if create_t_graph_response["status_info"]["status"] == "OK": print("{} creation success!".format(GRAPH_T)) print("Number of nodes: {}".format(create_t_graph_response["num_nodes"])) print("Number of edges: {}".format(create_t_graph_response["num_edges"])) else: print("{} creation failure: \n\t{}".format( GRAPH_T, create_t_graph_response["status_info"]["message"] )) print("Exiting...") sys.exit() # end graph_setup() ############################################################################ # # # Multiple Routing Solving Function # # # ############################################################################ def multiple_routing_example(): """Demonstrate creating and solving a graph using the MULTIPLE_ROUTING method (Traveling Salesman). """ print("\n===============================") print("MULTIPLE ROUTING SOLVER EXAMPLE") print("===============================\n") # Clear any related tables in case they already exist kinetica.clear_table(table_name=TABLE_GRAPH_T_MRSOLVED, options=OPTION_NO_DROP_ERROR) # Solve GRAPH_T for Multiple Routing using the given source node and # routing to the given destination nodes print("Solving {} using MULTIPLE_ROUTING solver type.".format(GRAPH_T)) source_node = "POINT(-73.98438262939453 40.76493835449219)" # node 2 destination_nodes = [ "POINT(-73.97122955322266 40.74700927734375)", # node 39 "POINT(-73.97740173339844 40.77263641357422)" # node 85 ] kinetica.solve_graph( graph_name = GRAPH_T, solver_type = "MULTIPLE_ROUTING", source_nodes = [source_node], destination_nodes = destination_nodes, solution_table = TABLE_GRAPH_T_MRSOLVED ) print("Cost for source node {} to visit destination nodes {}:".format(source_node, destination_nodes)) aggregate_records( kinetica, TABLE_GRAPH_T_MRSOLVED, ["SUM(SOLVERS_NODE_COSTS)"], ["Cost (in $)"], tblfmt = "psql" ) # end multiple_routing_example() def aggregate_records(database, table_name, column_names, column_headers = None, order_by = None, tblfmt = "grid"): """ Aggregate and tabulate a set of records from the given table, displaying the given columns with optional column headers and optionally sorted by the given ordering. Parameters: database (str) :class:`GPUdb` database connection object table_name (str) Name of the table whose records will be displayed column_names (list of str) Names of the columns or column aggregates whose values will be displayed column_headers (list of str) Header text to display at the top of the respective columns; should align with the columns specified in `column_names` order_by (list of str) Name(s) of the column(s) by which the results should be sorted, in the sequence they should be sorted; accepts `asc` & `desc` for forward/reverse sorting tblfmt (str) Output style, conforming to `tabulate` supported formats listed here: https://github.com/astanin/python-tabulate#table-format """ options = {'order_by': ','.join(order_by)} if order_by is not None else {} resp = database.aggregate_group_by( table_name, column_names, encoding = 'json', options = options ) if 'json_encoded_response' not in resp: print("[{}] Can't retrieve records: {}".format(resp['status_info']['status'], resp['status_info']['message'])) else: resp = json.loads(resp['json_encoded_response']) records = zip(*(resp['column_' + str(i)] for i in range(1, len(column_names) + 1))) headers = "keys" if column_headers is None else column_headers print(tabulate(records, headers=headers, tablefmt=tblfmt)) # end aggregate_records() if __name__ == '__main__': # Set up args parser = argparse.ArgumentParser(description='Run nyctaxi multiple route solve graph example.') parser.add_argument('--url', default='http://127.0.0.1:9191', help='Kinetica URL to run example against') parser.add_argument('--username', default='', help='Username of user to run example with') parser.add_argument('--password', default='', help='Password of user') parser.add_argument('--data_dir', default='./', help='Data file directory') args = parser.parse_args() # Establish connection with an instance of Kinetica, given a URL and credentials kinetica = gpudb.GPUdb(host = [args.url], username = args.username, password = args.password) # Execute defined functions table_setup(args.data_dir) graph_setup() multiple_routing_example()