package test1;

import java.util.Date;

import com.datastax.driver.core.*;

import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;

/*

* Simple Java client test to connect to trial cluster, create a time series data table, fill it, query it, and save it as csv for graphing.

*/

public class CassTest1 {

// the 3 node trial Cassandra test cluster Public IPs. These are dummy values.

static String n1PubIP = "01.23.45.678";

static String n2PubIP = "01.234.56.78";

static String n3PubIP = "01.23.456.78";

static String dcName = "hal_sydney"; // this is the DC name you used when created

static String user = "user";

static String password = "password";

public static void main(String[] args) {

long t1 = 0; // time each CQL operation, t1 is start time t2 is end time, time is t2-t1

long t2 = 0;

long time = 0;

Cluster.Builder clusterBuilder = Cluster.builder()

    .addContactPoints(

     n1PubIP, n2PubIP, n3PubIP // provide all 3 public IPs

    )

    .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dcName).build()) // your local data centre

    .withPort(9042)

    .withAuthProvider(new PlainTextAuthProvider(user, password));

Cluster cluster = null;

try {

cluster = clusterBuilder.build();

    Metadata metadata = cluster.getMetadata();

    System.out.printf("Connected to cluster: %s\n", metadata.getClusterName());

    for (Host host: metadata.getAllHosts()) {

        System.out.printf("Datacenter: %s; Host: %s; Rack: %s\n", host.getDatacenter(), host.getAddress(), host.getRack());

    }

    Session session = cluster.connect();                                          

    ResultSet rs;

    boolean createTable = true;

    if (createTable) {

     rs = session.execute("CREATE KEYSPACE IF NOT EXISTS hals WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3}");

     rs = session.execute("DROP TABLE IF EXISTS hals.sensordata");

     rs = session.execute("CREATE TABLE hals.sensordata(host text, metric text, time timestamp, value double, PRIMARY KEY ((host, metric), time) ) WITH CLUSTERING ORDER BY (time ASC)");

     System.out.println("Table hals.sensordata created!");

    }

    // Fill the table with some realistic sensor data. if createTable=false we just ADD data to the table

    double startValue = 100; // start value for random walk

    double nextValue = startValue; // next value in random walk, initially startValue

    int numHosts = 100; // how many host names to generate

    int toCreate = 1000; // how many times to pick a host name and create all metrics for it

    boolean usePrepared = false;

    PreparedStatement prepared = null;

    // prepare a prepared statement

    if (usePrepared)

     {

     System.out.println("Using PREPARED statements for INSERT");

     prepared = session.prepare("insert into hals.sensordata (host, metric, time, value) values (?, ?, ?, ?)");

     }

    t1 = System.currentTimeMillis();

    System.out.println("Creating data... iterations = " + toCreate);

    for (int r=1; r <= toCreate; r++) {

     long now = System.currentTimeMillis();

     Date date = new Date(now);

     // generate a random host name

     String hostname = "host" + (long)Math.round((Math.random() * numHosts));

     // do a random walk to produce realistic data

     double rand = Math.random();

     if (rand < 0.5)

     // 50% chance that value doesn't change

     ;

     else if (rand < 0.75)

     // 25% chance that value increases by 1

     nextValue++;

     else

     // 25% chance that value decreases by 1

     nextValue--;

     // never go negative

     if (nextValue < 0)

     nextValue = 0;

     // comparison of prepared vs. non-prepared statements

     if (usePrepared)  {

     session.execute(prepared.bind("'" + hostname + "'", "'m1'", date, nextValue));

     session.execute(prepared.bind("'" + hostname + "'", "'m2'", date, nextValue * 10));

     session.execute(prepared.bind("'" + hostname + "'", "'m3'", date, nextValue * 100));

     }

     else {

     // fake three metrics (m1, m2, m3) which are somehow related.

     rs = session.execute("insert into hals.sensordata (host, metric, time, value) values (" + "'" + hostname + "'" + ", " + "'m1'" + ", " + now + "," + (nextValue) + ");" );

     rs = session.execute("insert into hals.sensordata (host, metric, time, value) values (" + "'" + hostname + "'" + ", " + "'m2'" + ", " + now + "," + (nextValue * 10) + ");" );

     rs = session.execute("insert into hals.sensordata (host, metric, time, value) values (" + "'" + hostname + "'" + ", " + "'m3'" + ", " + now + "," + (nextValue * 100) + ");" );

     }

     }

    t2 = System.currentTimeMillis();

    System.out.println("Created rows = " + toCreate*3 + " in time = " + (t2-t1));

    // find the max value for a sample

    System.out.println("Getting max value for sample...");

    t1 = System.currentTimeMillis();

    rs = session.execute("select max(value) from hals.sensordata where host='host1' and metric='m1'");  

    t2 = System.currentTimeMillis();

    time = t2-t1;

    Row row = rs.one();

    System.out.println("Max value = " + row.toString() + " in time = " + time);  

    // get all the values for a sample

    System.out.println("Getting all rows for sample...");

    t1 = System.currentTimeMillis();

    rs = session.execute("select * from hals.sensordata where host='host1' and metric='m1'");

    for (Row rowN : rs) {

     System.out.println(rowN.toString());

    }

    t2 = System.currentTimeMillis();

    time = t2-t1;

    System.out.println("time = " + time);  

    // get all host/metric permutations

    System.out.println("Getting all host/metric permutations");

    t1 = System.currentTimeMillis();

    rs = session.execute("select distinct host, metric from hals.sensordata");

    for (Row rowN : rs) {

     System.out.println(rowN.toString());

    }

    t2 = System.currentTimeMillis();

    time = t2-t1;

    System.out.println("time = " + time);  

    // Note that SELECT * will return all results without limit (even though the driver might use multiple queries in the background).

    // To handle large result sets, you use a LIMIT clause in your CQL query, or use one of the techniques described in the paging documentation.

    System.out.println("Select ALL...");

    t1 = System.currentTimeMillis();

    rs = session.execute("select * from hals.sensordata");  

    System.out.println("Got rows (without fetching) = " + rs.getAvailableWithoutFetching());

    int i = 0;

    long numBytes = 0;

    // example use of the data: count rows and total bytes returned.

    for (Row rowN : rs)

    {

     i++;

     numBytes += rowN.toString().length();

    }

    t2 = System.currentTimeMillis();

    time = t2-t1;

    System.out.println("Returned rows = " + i + ", total bytes = " + numBytes + ", in time = " + time);

} finally {

if (cluster != null) cluster.close();    

}

}

}