Initial commit

{\rtf1\ansi\ansicpg1252\cocoartf1038\cocoasubrtf350

{\fonttbl\f0\fswiss\fcharset0 Helvetica;}

{\colortbl;\red255\green255\blue255;}

\margl1440\margr1440\vieww9000\viewh8400\viewkind0

\pard\tx720\tx1440\tx2160\tx2880\tx3600\tx4320\tx5040\tx5760\tx6480\tx7200\tx7920\tx8640\ql\qnatural\pardirnatural

\f0\fs24 \cf0 No work files for this lesson...}

 No newline at end of file

{\rtf1\ansi\ansicpg1252\cocoartf1038\cocoasubrtf350

{\fonttbl\f0\fswiss\fcharset0 Helvetica;}

{\colortbl;\red255\green255\blue255;}

\margl1440\margr1440\vieww9000\viewh8400\viewkind0

\pard\tx720\tx1440\tx2160\tx2880\tx3600\tx4320\tx5040\tx5760\tx6480\tx7200\tx7920\tx8640\ql\qnatural\pardirnatural

\f0\fs24 \cf0 No work files for this lesson...}

 No newline at end of file

package com.oreilly.accumulotraining;

import java.io.File;

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.Iterator;

import java.util.concurrent.TimeUnit;

import java.util.logging.Level;

import java.util.logging.Logger;

import org.apache.accumulo.core.client.AccumuloException;

import org.apache.accumulo.core.client.AccumuloSecurityException;

import org.apache.accumulo.core.client.BatchWriter;

import org.apache.accumulo.core.client.BatchWriterConfig;

import org.apache.accumulo.core.client.Connector;

import org.apache.accumulo.core.client.Durability;

import org.apache.accumulo.core.client.Instance;

import org.apache.accumulo.core.client.MutationsRejectedException;

import org.apache.accumulo.core.client.TableExistsException;

import org.apache.accumulo.core.client.TableNotFoundException;

import org.apache.accumulo.core.client.ZooKeeperInstance;

import org.apache.accumulo.core.client.security.tokens.PasswordToken;

import org.apache.accumulo.core.data.Mutation;

import org.apache.accumulo.core.data.Value;

import org.apache.commons.csv.CSVFormat;

import org.apache.commons.csv.CSVParser;

import org.apache.commons.csv.CSVRecord;

public class IngestClient {

	private static final Logger logger = Logger.getLogger(IngestClient.class.getName());

	private static final int COUNTRY = 0;

	private static final int COMMODITY = 1;

	private static final int YEAR = 2;

	private static final int UNIT = 3;

	private static final int QUANTITY = 4;

	private static final int FOOTNOTES = 5;

	public static void run(

			String instanceName, 

			String zookeepers, 

			String username, 

			String password, 

			String table, 

			String filename) {

		try {

			System.out.println("connecting to accumulo ...");

			Instance inst = new ZooKeeperInstance(instanceName, zookeepers);

			System.out.println("got instance");

			Connector conn = inst.getConnector(username, new PasswordToken(password));

			System.out.println("got connector");

			if(!conn.tableOperations().exists(table)) {

				System.out.println("creating table " + table);

				conn.tableOperations().create(table);

			}

			BatchWriterConfig config = new BatchWriterConfig();

			config.setMaxLatency(1, TimeUnit.SECONDS);

			config.setMaxMemory(10240);

			config.setDurability(Durability.DEFAULT);

			config.setMaxWriteThreads(10);

			BatchWriter writer = conn.createBatchWriter(table, config);

			int written = 0;

			File csvData = new File(filename);

			System.out.println("parsing file");

			CSVParser parser = CSVParser.parse(csvData, Charset.defaultCharset(), CSVFormat.EXCEL);

			System.out.println("writing data from file " + filename + " ...");

			Iterator<CSVRecord> records = parser.getRecords().iterator();

			records.next(); // skip header

			while(records.hasNext()) {

				CSVRecord csvRecord = records.next();

				if(csvRecord.size() < 6) {

					continue;

				}

				String energyType = parseEnergyType(csvRecord.get(COMMODITY));

				Mutation m = new Mutation(csvRecord.get(COUNTRY));

				m.put(

						energyType, 

						csvRecord.get(YEAR), 

						new Value(csvRecord.get(QUANTITY).getBytes()));

				writer.addMutation(m);

				written++;

			}

			writer.close();

			parser.close();

			System.out.println("wrote " + written + " records");

		} catch (MutationsRejectedException ex) {

			// see Error Handling Example

		} catch (AccumuloException | AccumuloSecurityException | TableNotFoundException | TableExistsException | IOException ex) {

			logger.log(Level.SEVERE, ex.getLocalizedMessage(), ex);

		}

	}	

	private static String parseEnergyType(String line) {

		String[] parts = line.split("\\s+");

		if(parts.length > 3)

			return parts[3];

		return line;

	}

}

package com.oreilly.accumulotraining;

import java.util.Map;

import java.util.logging.Level;

import java.util.logging.Logger;

import org.apache.accumulo.core.client.AccumuloException;

import org.apache.accumulo.core.client.AccumuloSecurityException;

import org.apache.accumulo.core.client.Connector;

import org.apache.accumulo.core.client.Instance;

import org.apache.accumulo.core.client.Scanner;

import org.apache.accumulo.core.client.TableNotFoundException;

import org.apache.accumulo.core.client.ZooKeeperInstance;

import org.apache.accumulo.core.client.security.tokens.PasswordToken;

import org.apache.accumulo.core.data.Key;

import org.apache.accumulo.core.data.Range;

import org.apache.accumulo.core.data.Value;

import org.apache.accumulo.core.security.Authorizations;

import org.apache.hadoop.io.Text;

public class ScanClient {

	private static final Logger logger = Logger.getLogger(ScanClient.class.getName());

	public static void run(

			String instanceName,

			String zookeepers,

			String username,

			String password,

			String table,

			String row,

			String columnFamily,

			String columnQualifier) {

		try {

			Instance inst = new ZooKeeperInstance(instanceName, zookeepers);

			Connector conn = inst.getConnector(username, new PasswordToken(password));

			Scanner scanner = conn.createScanner(table, Authorizations.EMPTY);

			if(row != null) {

				scanner.setRange(Range.exact(row));

			}

			if(columnFamily != null) {

				if(columnQualifier != null) {

					scanner.fetchColumn(new Text(columnFamily), new Text(columnQualifier));

				}

				else {

					scanner.fetchColumnFamily(new Text(columnFamily));

				}

			}

			for(Map.Entry<Key, Value> e : scanner) {

				System.out.println(

						e.getKey().getRow().toString() + " " +

						e.getKey().getColumnFamily().toString() + " " +

						e.getKey().getColumnQualifier().toString() + "\t" +

						new String(e.getValue().get()));

			}

		} catch (AccumuloException | AccumuloSecurityException | TableNotFoundException ex) {

			logger.log(Level.SEVERE, ex.getLocalizedMessage(), ex);

		}

	}

}

package com.oreilly.accumulotraining;

import java.util.Map;

import java.util.SortedSet;

import java.util.logging.Level;

import java.util.logging.Logger;

import org.apache.accumulo.core.client.AccumuloException;

import org.apache.accumulo.core.client.AccumuloSecurityException;

import org.apache.accumulo.core.client.Connector;

import org.apache.accumulo.core.client.Instance;

import org.apache.accumulo.core.client.TableExistsException;

import org.apache.accumulo.core.client.TableNotFoundException;

import org.apache.accumulo.core.client.ZooKeeperInstance;

import org.apache.accumulo.core.client.admin.NewTableConfiguration;

import org.apache.accumulo.core.client.admin.TableOperations;

import org.apache.accumulo.core.client.admin.TimeType;

import org.apache.accumulo.core.client.security.tokens.PasswordToken;

import org.apache.hadoop.io.Text;

/**

 *

 */

public class TableCode {

	public static void main(String[] args) {

		Map<String, String> properties = null;

		Instance instance = new ZooKeeperInstance("instance name", "zookeeper servers");

		try {

			Connector connector = instance.getConnector("username", new PasswordToken("password"));

			TableOperations ops = connector.tableOperations();

			SortedSet<String> tables = ops.list();

			for(String table : tables) {

				System.out.println(table);

			}			

			if (!ops.exists("table")) {

				ops.create("table");

				NewTableConfiguration tableConfiguration = new NewTableConfiguration();

				tableConfiguration.setProperties(properties);

				tableConfiguration.setTimeType(TimeType.MILLIS);

				ops.create("table", tableConfiguration);

			}

			ops.delete("table");

			Text startKey = null;

			Text endKey = null;

			ops.deleteRows("table", startKey, endKey);

			boolean wait = false;

			ops.flush("table", startKey, endKey, wait);

			boolean flush = true;

			ops.compact("table", startKey, endKey, flush, wait);

		} catch (AccumuloException | AccumuloSecurityException | TableNotFoundException | TableExistsException ex) {

			Logger.getLogger(TableCode.class.getName()).log(Level.SEVERE, null, ex);

		}

	}

}