chbatey

@Configuration
@EnableWebSecurity
public class SecurityConfig extends WebSecurityConfigurerAdapter {

    private static final Logger LOGGER = LoggerFactory.getLogger(SecurityConfig.class);

    @Inject
    private UserDetailsService userDetailsService;

    @Inject

chbatey

val conf = new SparkConf().set("spark.cassandra.connection.host", cassandraHost)
val sc = new SparkContext(sparkMaster, "CassandraSchemaMigration", conf)

val events_by_customer = sc.cassandraTable("keyspace", "customer_events")

events_by_customer.saveToCassandra("keyspace", "customer_events_by_staff", 
   SomeColumns("customer_id", "time", "id", "event_type", "staff_name", "staff_title", 
   "store_location", "store_name", "store_type"))

chbatey

CREATE TABLE IF NOT EXISTS customer_events_by_staff(  
  customer_id text,   
  time timestamp, 
  id uuid,  
  event_type text, 
  store_name text, 
  store_type text, 
  store_location text, 
  staff_name text, 
  staff_title text,  

chbatey

CREATE TABLE IF NOT EXISTS customer_events(
  customer_id text,   
  time timestamp, 
  id uuid,  
  event_type text, 
  store_name text, 
  store_type text, 
  store_location text, 
  staff_name text, 
  staff_title text,  

chbatey

val customerEvents = new JdbcRDD(sc, () => { DriverManager.getConnection(mysqlJdbcString)},
    "select * from customer_events ce, staff, store where ce.store = store.store_name and ce.staff = staff.name " +
        "and ce.id >= ? and ce.id <= ?", startingId, highestId, numberOfPartitions,
    (r: ResultSet) => {
      (r.getString("customer"),
        r.getTimestamp("time"),
        UUID.randomUUID(),
        r.getString("event_type"),
        r.getString("store_name"),
        r.getString("location"),

chbatey

  CassandraConnector(conf).withSessionDo { session =>
    session.execute("CREATE KEYSPACE IF NOT EXISTS test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1 }")
    session.execute("CREATE TABLE IF NOT EXISTS test.customer_events(  customer_id text,   time timestamp, id uuid,  event_type text, " +
      "store_name text, store_type text, store_location text, staff_name text, staff_title text,  PRIMARY KEY ((customer_id), time, id))")
  }

chbatey

import com.datastax.spark.connector._
import com.datastax.spark.connector.cql.CassandraConnector
import org.apache.spark._
import org.apache.spark.rdd.JdbcRDD

chbatey

val conf = new SparkConf().set("spark.cassandra.connection.host", "127.0.0.1")
val sc = new SparkContext("local[2]", "MigrateMySQLToCassandra", conf)
val mysqlJdbcString: String = s"jdbc:mysql://192.168.10.11/customer_events?user=root&password=password"
Class.forName("com.mysql.jdbc.Driver").newInstance

chbatey

insert into staff(name, favourite_colour, job_title) values ('Charlie', 'Blue', 'Awesome Marketer');
insert into store(store_name, location, store_type) values ('ChrisBatey.com', 'US', 'WEB');
insert into customer_events(customer, time, event_type, store, staff) values ('chbatey', now(), 'BUY_MOVIE', 'ChrisBatey.com', 'Charlie');
insert into customer_events(customer, time, event_type, store, staff) values ('chbatey', now(), 'WATCH_MOVIE', 'ChrisBatey.com', 'Charlie');

chbatey

create table store(
  store_name varchar(32) primary key, 
  location varchar(32), 
  store_type varchar(10));

create table staff(
  name varchar(32) 
  primary key, 
  favourite_colour varchar(32), 
  job_title varchar(32));