· 9 years ago · Aug 31, 2017, 12:18 AM
1package test1;
2
3import java.util.Date;
4
5import com.datastax.driver.core.*;
6import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
7
8/*
9 * 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.
10 */
11
12public class CassTest1 {
13 // the 3 node trial Cassandra test cluster Public IPs. These are dummy values.
14 static String n1PubIP = "01.23.45.678";
15 static String n2PubIP = "01.234.56.78";
16 static String n3PubIP = "01.23.456.78";
17 static String dcName = "hal_sydney"; // this is the DC name you used when created
18 static String user = "user";
19 static String password = "password";
20
21 public static void main(String[] args) {
22
23 long t1 = 0; // time each CQL operation, t1 is start time t2 is end time, time is t2-t1
24 long t2 = 0;
25 long time = 0;
26
27 Cluster.Builder clusterBuilder = Cluster.builder()
28 .addContactPoints(
29 n1PubIP, n2PubIP, n3PubIP // provide all 3 public IPs
30 )
31 .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dcName).build()) // your local data centre
32 .withPort(9042)
33 .withAuthProvider(new PlainTextAuthProvider(user, password));
34
35 Cluster cluster = null;
36 try {
37 cluster = clusterBuilder.build();
38
39 Metadata metadata = cluster.getMetadata();
40 System.out.printf("Connected to cluster: %s\n", metadata.getClusterName());
41
42 for (Host host: metadata.getAllHosts()) {
43 System.out.printf("Datacenter: %s; Host: %s; Rack: %s\n", host.getDatacenter(), host.getAddress(), host.getRack());
44 }
45
46 Session session = cluster.connect();
47
48 ResultSet rs;
49 boolean createTable = true;
50 if (createTable) {
51 rs = session.execute("CREATE KEYSPACE IF NOT EXISTS hals WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3}");
52 rs = session.execute("DROP TABLE IF EXISTS hals.sensordata");
53 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)");
54 System.out.println("Table hals.sensordata created!");
55 }
56
57 // Fill the table with some realistic sensor data. if createTable=false we just ADD data to the table
58
59 double startValue = 100; // start value for random walk
60 double nextValue = startValue; // next value in random walk, initially startValue
61 int numHosts = 100; // how many host names to generate
62 int toCreate = 1000; // how many times to pick a host name and create all metrics for it
63
64 boolean usePrepared = false;
65 PreparedStatement prepared = null;
66 // prepare a prepared statement
67 if (usePrepared)
68 {
69 System.out.println("Using PREPARED statements for INSERT");
70 prepared = session.prepare("insert into hals.sensordata (host, metric, time, value) values (?, ?, ?, ?)");
71 }
72
73 t1 = System.currentTimeMillis();
74 System.out.println("Creating data... iterations = " + toCreate);
75 for (int r=1; r <= toCreate; r++) {
76 long now = System.currentTimeMillis();
77 Date date = new Date(now);
78 // generate a random host name
79 String hostname = "host" + (long)Math.round((Math.random() * numHosts));
80 // do a random walk to produce realistic data
81 double rand = Math.random();
82 if (rand < 0.5)
83 // 50% chance that value doesn't change
84 ;
85 else if (rand < 0.75)
86 // 25% chance that value increases by 1
87 nextValue++;
88 else
89 // 25% chance that value decreases by 1
90 nextValue--;
91 // never go negative
92 if (nextValue < 0)
93 nextValue = 0;
94
95 // comparison of prepared vs. non-prepared statements
96 if (usePrepared) {
97 session.execute(prepared.bind("'" + hostname + "'", "'m1'", date, nextValue));
98 session.execute(prepared.bind("'" + hostname + "'", "'m2'", date, nextValue * 10));
99 session.execute(prepared.bind("'" + hostname + "'", "'m3'", date, nextValue * 100));
100 }
101 else {
102 // fake three metrics (m1, m2, m3) which are somehow related.
103 rs = session.execute("insert into hals.sensordata (host, metric, time, value) values (" + "'" + hostname + "'" + ", " + "'m1'" + ", " + now + "," + (nextValue) + ");" );
104 rs = session.execute("insert into hals.sensordata (host, metric, time, value) values (" + "'" + hostname + "'" + ", " + "'m2'" + ", " + now + "," + (nextValue * 10) + ");" );
105 rs = session.execute("insert into hals.sensordata (host, metric, time, value) values (" + "'" + hostname + "'" + ", " + "'m3'" + ", " + now + "," + (nextValue * 100) + ");" );
106 }
107 }
108
109 t2 = System.currentTimeMillis();
110 System.out.println("Created rows = " + toCreate*3 + " in time = " + (t2-t1));
111
112 // find the max value for a sample
113 System.out.println("Getting max value for sample...");
114 t1 = System.currentTimeMillis();
115 rs = session.execute("select max(value) from hals.sensordata where host='host1' and metric='m1'");
116 t2 = System.currentTimeMillis();
117 time = t2-t1;
118
119 Row row = rs.one();
120 System.out.println("Max value = " + row.toString() + " in time = " + time);
121
122 // get all the values for a sample
123 System.out.println("Getting all rows for sample...");
124 t1 = System.currentTimeMillis();
125 rs = session.execute("select * from hals.sensordata where host='host1' and metric='m1'");
126 for (Row rowN : rs) {
127 System.out.println(rowN.toString());
128 }
129 t2 = System.currentTimeMillis();
130 time = t2-t1;
131 System.out.println("time = " + time);
132
133 // get all host/metric permutations
134 System.out.println("Getting all host/metric permutations");
135 t1 = System.currentTimeMillis();
136 rs = session.execute("select distinct host, metric from hals.sensordata");
137 for (Row rowN : rs) {
138 System.out.println(rowN.toString());
139 }
140 t2 = System.currentTimeMillis();
141 time = t2-t1;
142 System.out.println("time = " + time);
143
144 // Note that SELECT * will return all results without limit (even though the driver might use multiple queries in the background).
145 // 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.
146 System.out.println("Select ALL...");
147 t1 = System.currentTimeMillis();
148 rs = session.execute("select * from hals.sensordata");
149 System.out.println("Got rows (without fetching) = " + rs.getAvailableWithoutFetching());
150 int i = 0;
151 long numBytes = 0;
152 // example use of the data: count rows and total bytes returned.
153 for (Row rowN : rs)
154 {
155 i++;
156 numBytes += rowN.toString().length();
157 }
158 t2 = System.currentTimeMillis();
159 time = t2-t1;
160 System.out.println("Returned rows = " + i + ", total bytes = " + numBytes + ", in time = " + time);
161
162 } finally {
163 if (cluster != null) cluster.close();
164 }
165 }
166}