Loading CSV data into a Grakn keyspace using the Node.js Client

loadCsvIntoGrakn.js

const Grakn = require("grakn");
const fs = require("fs");
const papa = require("papaparse");

const inputs = [
  { dataPath: "./data/companies", template: companyTemplate },
  { dataPath: "./data/people", template: personTemplate },
  { dataPath: "./data/contracts", template: contractTemplate },
  { dataPath: "./data/calls", template: callTemplate }
];

// Go
buildPhoneCallGraph(inputs);

async function buildPhoneCallGraph() {
  const grakn = new Grakn("localhost:48555");
  const session = grakn.session("phone_calls");

  for (input of inputs) {
    console.log("Loading from [" + input.dataPath + "] into Grakn ...");
    await loadDataIntoGrakn(input, session);
  }

  session.close();
}

async function loadDataIntoGrakn(input, session) {
  const items = await parseDataToObjects(input);

  for (item of items) {
    let tx;
    tx = await session.transaction(Grakn.txType.WRITE);

    const graqlInsertQuery = input.template(item);
    console.log("Executing Graql Query: " + graqlInsertQuery);
    await tx.query(graqlInsertQuery);
    await tx.commit();
  }

  console.log(
    `\nInserted ${items.length} items from [${input.dataPath}] into Grakn.\n`
  );
}

async function loadDataIntoGrakn(input, session) {
  const items = await parseDataToObjects(input);

  for (item of items) {
    const tx = await session.transaction(Grakn.txType.WRITE);

    const graqlInsertQuery = input.template(item);
    console.log("Executing Graql Query: " + graqlInsertQuery);
    tx.query(graqlInsertQuery);
    tx.commit();
  }

  console.log(
    `\nInserted ${items.length} items from [${input.dataPath}] into Grakn.\n`
  );
}

function companyTemplate(company) {
  return `insert $company isa company has name "${company.name}";`;
}

function personTemplate(person) {
  const { first_name, last_name, phone_number, city, age } = person;
  // insert person
  let graqlInsertQuery = `insert $person isa person has phone-number "${phone_number}"`;
  const isNotCustomer = first_name === "";
  if (isNotCustomer) {
    // person is not a customer
    graqlInsertQuery += " has is-customer false";
  } else {
    // person is a customer
    graqlInsertQuery += ` has is-customer true`;
    graqlInsertQuery += ` has first-name "${first_name}"`;
    graqlInsertQuery += ` has last-name "${last_name}"`;
    graqlInsertQuery += ` has city "${city}"`;
    graqlInsertQuery += ` has age ${age}`;
  }

  graqlInsertQuery += ";";
  return graqlInsertQuery;
}

function contractTemplate(contract) {
  const { company_name, person_id } = contract;
  // match company
  let graqlInsertQuery = `match $company isa company has name "${company_name}"; `;
  // match person
  graqlInsertQuery += `$customer isa person has phone-number "${person_id}"; `;
  // insert contract
  graqlInsertQuery +=
    "insert (provider: $company, customer: $customer) isa contract;";
  return graqlInsertQuery;
}

function callTemplate(call) {
  const { caller_id, callee_id, started_at, duration } = call;
  // match caller
  let graqlInsertQuery = `match $caller isa person has phone-number "${caller_id}"; `;
  // match callee
  graqlInsertQuery += `$callee isa person has phone-number "${callee_id}"; `;
  // insert call
  graqlInsertQuery += "insert $call(caller: $caller, callee: $callee) isa call; " +
    `$call has started-at ${started_at}; $call has duration ${duration};`;
  return graqlInsertQuery;
}

function parseDataToObjects(input) {
  const items = [];
  return new Promise(function(resolve, reject) {
    papa.parse(
      fs.createReadStream(input.dataPath + ".csv"),
      {
        header: true, // a Papaparse config option
        step: function(results, parser) {
          items.push(results.data[0]);
        },
        complete: function() {
          resolve(items);
        }
      }
    );
  });
}