val transformer = CqlRowReader.columnListMapper[Employee]("deptid", "empid", "first_name", "last_name")

      import transformer._

      val cas = CasBuilder.cql3.withColumnFamily("cql3_test", "emp_read_test")

      val casrdd = sc.cql3Cassandra[Employee](cas)

      val result = casrdd.collect().toList

      result must have length (5)

      result should contain(Employee(20, 105, "jack", "carpenter"))

      result should contain(Employee(20, 106, "john", "grumpy"))

import org.apache.spark._ 

import com.datastax.spark.connector._

val conf = new SparkConf(true).set("spark.cassandra.connection.host", "ip") 

val sc = new SparkContext("spark://ip:7077", "test", conf)

val testRDD = sc.cassandraTable("testKs", "kv") 

println(testRDD.count) 

println(testRDD.first)

println(testRDD.map(_.getInt("value")).sum) 

val col = sc.parallelize(Seq(("key3", 3), ("key4", 4))) 

col.saveToCassandra("testKs", "kv", SomeColumns("key", "value"))      

scala> case class Member(id: Int, name: String, dept: Int)

scala> val members = sc.parallelize(List(Member(1,"john", 100), Member(2,"samuel", 200), Member(3,"ethan", 200)))

scala> case class Department(id: Int, name: String)

scala> val departments = sc.parallelize(List(Department(100, "server"), Department(200, "client")))

scala> val groupedMembers = members.groupBy(x => x.dept)

groupedMembers: org.apache.spark.rdd.RDD[(Int, Iterable[Member])] = ShuffledRDD[128] at groupBy at <console>:29

scala> groupedMembers.foreach(println)

(100,CompactBuffer(Member(1,john,100)))

(200,CompactBuffer(Member(2,samuel,200), Member(3,ethan,200)))

scala> val groupedDepartments = departments.groupBy(x => x.id)

groupedDepartments: org.apache.spark.rdd.RDD[(Int, Iterable[Department])] = ShuffledRDD[130] at groupBy at <console>:29

scala> groupedDepartments.foreach(println)

(200,CompactBuffer(Department(200,client)))

(100,CompactBuffer(Department(100,server)))

scala> groupedMembers.join(groupedDepartments).foreach(println)

(200,(CompactBuffer(Member(2,samuel,200), Member(3,ethan,200)),CompactBuffer(Department(200,client))))

(100,(CompactBuffer(Member(1,john,100)),CompactBuffer(Department(100,server))))

scala> val df = sc.parallelize(Seq((1, "samuel"), (2, "jack"), (3, "jonathan"))).toDF("id", "name")

df: org.apache.spark.sql.DataFrame = [id: int, column: string]

scala> df("name")

res11: org.apache.spark.sql.Column = name

scala> df.show

+---+--------+

| id|    name|

+---+--------+

|  1|  samuel|

|  2|    jack|

|  3|jonathan|

+---+--------+

scala> df.select("id").show

+---+

| id|

+---+

|  1|

|  2|

|  3|

+---+

scala> df.select("name").show

+--------+

|    name|

+--------+

|  samuel|

|    jack|

|jonathan|

+--------+

scala> df.where(df("name").contains("j")).show

+---+--------+

| id|    name|

+---+--------+

|  2|    jack|

|  3|jonathan|

+---+--------+

scala> df.where(col("id") between (1,2)).show

+---+------+

| id|  name|

+---+------+

|  1|samuel|

|  2|  jack|

+---+------+

scala> df.select("name").where(col("id") between (1,2)).show

+------+

|  name|

+------+

|samuel|

|  jack|

+------+

scala> df.withColumn("xx", col("id")).show

+---+--------+---+

| id|    name| xx|

+---+--------+---+

|  1|  samuel|  1|

|  2|    jack|  2|

|  3|jonathan|  3|

+---+--------+---+

scala> df.drop(col("xxx"))

res56: org.apache.spark.sql.DataFrame = [id: int, name: string]

scala> df.show

+---+--------+

| id|    name|

+---+--------+

|  1|  samuel|

|  2|    jack|

|  3|jonathan|

+---+--------+

scala> import org.apache.spark.sql.functions.udf

import org.apache.spark.sql.functions.udf

scala> case class Member(id:Int, name:String)

defined class Member

scala> case class Member(id:Int, name:String)

defined class Member

scala> val mudf=udf((id:Int, name:String) => Member(1*100, name))

mudf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function2>,StructType(StructField(id,IntegerType,false), StructField(name,StringType,true)),Some(List(IntegerType, StringType)))

scala> val df1 = df.withColumn("newColumn", mudf($"id", $"name"))

df1: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]

scala> df1.show

+---+--------+--------------+

| id|    name|     newColumn|

+---+--------+--------------+

|  1|  samuel|  [100,samuel]|

|  2|    jack|    [100,jack]|

|  3|jonathan|[100,jonathan]|

+---+--------+--------------+

rom log4j:ERROR Failed to flush writer,

java.io.IOException: 장치에 남은 공간이 없음

at java.io.FileOutputStream.writeBytes(Native Method)

at java.io.FileOutputStream.write(FileOutputStream.java:345)

at sun.nio.cs.StreamEncoder.writeBytes(StreamEncoder.java:221)

at sun.nio.cs.StreamEncoder.implFlushBuffer(StreamEncoder.java:291)

at sun.nio.cs.StreamEncoder.implFlush(StreamEncoder.java:295)

at sun.nio.cs.StreamEncoder.flush(StreamEncoder.java:141)

at java.io.OutputStreamWriter.flush(OutputStreamWriter.java:229)

at org.apache.log4j.helpers.QuietWriter.flush(QuietWriter.java:59)

at org.apache.log4j.WriterAppender.subAppend(WriterAppender.java:324)

at org.apache.log4j.DailyRollingFileAppender.subAppend(DailyRollingFileAppender.java:369)

at org.apache.log4j.WriterAppender.append(WriterAppender.java:162)

at org.apache.log4j.AppenderSkeleton.doAppend(AppenderSkeleton.java:251)

at org.apache.log4j.helpers.AppenderAttachableImpl.appendLoopOnAppenders(AppenderAttachableImpl.java:66)

at org.apache.log4j.Category.callAppenders(Category.java:206)

at org.apache.log4j.Category.forcedLog(Category.java:391)

at org.apache.log4j.Category.log(Category.java:856)

at org.apache.commons.logging.impl.Log4JLogger.info(Log4JLogger.java:176)

at org.apache.hadoop.hive.ql.session.SessionState.createPath(SessionState.java:641)

at org.apache.hadoop.hive.ql.session.SessionState.createSessionDirs(SessionState.java:578)

at org.apache.hadoop.hive.ql.session.SessionState.start(SessionState.java:508)

at org.apache.hadoop.hive.cli.CliDriver.run(CliDriver.java:671)

at org.apache.hadoop.hive.cli.CliDriver.main(CliDriver.java:615)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.lang.reflect.Method.invoke(Method.java:606)

at org.apache.hadoop.util.RunJar.run(RunJar.java:221)

at org.apache.hadoop.util.RunJar.main(RunJar.java:136)

import java.util.concurrent.Callable;

import java.util.concurrent.FutureTask;

...

                 String text = "1";

Callable<Integer> task = new Callable<Integer>() {

@Override

public Integer call() throws Exception {

return Integer.parseInt(text);

}

};

FutureTask<Integer> futureTask = new FutureTask<Integer>(task);

futureTask.run();

try {

System.out.println(futureTask.get());

} catch (Exception e) {

e.printStackTrace();

}

import java.util.concurrent.Callable;

import java.util.concurrent.FutureTask;

...

String text = "1";

FutureTask<Integer> futureTask = new FutureTask<Integer>(() -> {

return Integer.parseInt(text);

});

futureTask.run();

try {

System.out.println(futureTask.get());

} catch (Exception e) {

e.printStackTrace();

}

@FunctionalInterface

public interface Callable<V> {

    /**

     * Computes a result, or throws an exception if unable to do so.

     *

     * @return computed result

     * @throws Exception if unable to compute a result

     */

    V call() throws Exception;

}

@FunctionalInterface

public interface Runnable {

    /**

     * When an object implementing interface <code>Runnable</code> is used

     * to create a thread, starting the thread causes the object's

     * <code>run</code> method to be called in that separately executing

     * thread.

     * <p>

     * The general contract of the method <code>run</code> is that it may

     * take any action whatsoever.

     *

     * @see     java.lang.Thread#run()

     */

    public abstract void run();

}

Runnable runabble = new Runnable() {

  @Override

  public void run() {

  System.out.println("Run!!");

  }

};

Thread th = new Thread(runabble);

th.start();

   Thread.sleep(1000);

                Thread th = new Thread(() -> {

System.out.println("Run!!");

});

th.start();

              Thread.sleep(1000);

@FunctionalInterface

interface BindCallable {

public List<String> bind();

}

class Bind {

  public List<String> process(BindCallable callable){

    return callable.bind();

  }

}

Bind bind = new Bind();

List<String> result = bind.process(() -> {

final String a = "a";

final String b = "b";

List<String> list = Lists.newArrayList();

list.add(a); list.add(b);

return list;

});

System.out.println(result);

These were Twitter engineers who had once worked at Google: John Sirois, Travis Crawford, and Bill Farner. They told Hindman that they missed Borg, and that Mesos seemed like the perfect way to rebuild it.

He and his fellow engineers continued to run Mesos as an open source software project, but at Twitter, he also worked to move the platform into the company’s data center and fashion something very similar to Google Borg.

라이버트 CEO는 트위터의 엔지니어링을 이끌었던 인물이다. 그는 트위터의 신규 서비스에 모두 메소스를 사용하도록 했다. 트위터를 떠난 후 에어비앤비에서 메소스 상에 분석 환경을 구축했다. ETL 관리와 스케줄리을 위한 크로노스 아파치 메소스 프레임워크의 주요 저자기도 하다.

• Repeated Serialization - A tuple is represented as a collection of plain old Java objects. The Java objects are converted into byte arrays using either Kryo or Java serialization. The byte arrays are again serialized when included in a protocol buffers object used for data exchange between stream managers and Heron instances.
• Eager Deserialization - The stream manager is responsible for receiving and routing protocol buffer tuples. When it receives a tuple, it eagerly deserializes it into a C++ object.
• Immutability - To simplify the implementation and reasoning, stream manager does not reuse any protobuf objects. For each message received, it uses the malloc allocator to allocate a protobuf object, which it then releases back to the allocator once the operation is completed. Instead of modifying the protobuf in place, it copies the contents to a newly allocated message, makes the modification on the new message and releases the old one.

• ~17% of the CPU is used to create/delete a protobuf object from memory allocator (not including those protobuf objects allocated on stack).
• ~15% of the CPU is used to copy a new protobuf object instead of updating one in place.
• ~18% of the CPU is used to eagerly deserialize a protobuf message, despite the fact that eager deserialization is not needed; instead we could just handle the byte array.

• Added a separate memory pool for each type of protobuf message thereby reducing the expensive cost to create/delete a protobuf message.
• Changed an internal data structure that caches tuples from std::list to std::deque to facilitate preallocation of protobuf messages.
• Optimized away the code that was duplicating the protobuf message to do in-place update whenever possible.
• When a stream manager receives a message from another stream manager, instead of eagerly deserializing the inner tuple message, it now transfers the underlying serialized byte array directly to the instance.

ClassPathBeanDefinitionScanner scanner = new PlayClassPathBeanDefinitionScanner(applicationContext);

String scanBasePackage = ..

scanner.scan(scanBasePackage.split(","));

<spring 4>

https://github.com/spring-projects/spring-framework/blob/5b98a54c9b9f8c2f4332734ee23cd483b7df0d22/spring-context/src/main/java/org/springframework/context/annotation/ConfigurationClassParser.java#L659

/**

 ./spark-shell --packages datastax:spark-cassandra-connector:1.6.0-s_2.11

 ./spark-shell --conf "spark.driver.extraJavaOptions=-Dhttp.proxyHost=<proxyHost> -Dhttp.proxyPort=<proxyPort> -Dhttps.proxyHost=<proxyHost> -Dhttps.proxyPort=<proxyPort>" --packages datastax:spark-cassandra-connector:1.6.0-s_2.11