카산드라 스파크 커넥터 및 필터링 데이터

Question

내가 스파크 1.3.1을 사용하고 내가

같은 카산드라에

val sc = new SparkContext(conf) 
val rdd = sc.cassandraTable("foo", "bar") 
val date = DateTime.now().minusHours(1) 
val rdd2 = rdd.filter(r => r.getDate("date").after(date.toDate)) 
println(rdd2.count()) 
sc.stop() 

이 프로그램은, 아주 긴 시간 동안 실행 메시지를 출력 데이터를 필터링 할 수있는 작은 프로그램을 작성했습니다 내가 프로그램을 종료하고

val date = DateTime.now().minusHours(1) 
val rdd2 = rdd.filter(r => r.getDate("date").after(date.toDate)) 

에 코드를 변경하는 경우

16/09/01 21:10:31 INFO Executor: Running task 46.0 in stage 0.0 (TID 46) 
16/09/01 21:10:31 INFO TaskSetManager: Finished task 42.0 in stage 0.0 (TID 42) in 20790 ms on localhost (43/1350) 

는 아직 잘 모르겠지만 아주 동안 실행 프로그램은 항상 메모리에 전체 카산드라 테이블을로드하려고 (또는 완전히 스캔하려고) 만 다음 필터를 적용 할 것 같은

6/09/01 21:14:01 INFO Executor: Running task 8.0 in stage 0.0 (TID 8) 
16/09/01 21:14:01 INFO TaskSetManager: Finished task 4.0 in stage 0.0 (TID 4) in 19395 ms on localhost (5/1350) 

같은 메시지와 NG 시간 그래서 보인다. 그게 나에게 극도로 비효율적 인 것 같아.

스파크가 전체 cassandra 테이블을로드하거나 (완전히 스캔하여) RDD에 필터를 적용하지 않도록이 코드를 더 잘 작성하려면 어떻게해야합니까?

val rdd = sc.cassandraTable("foo", "bar") 
val date = DateTime.now().minusDays(30) 
rdd.filter(r => r.getDate("date").after(date.toDate)).count // Count Filtered RDD 

코드

의

Answer 1

첫 번째 작품은 그래서 조심. RDD는 불변이므로 필터를 적용 할 때 함수를 적용한 RDD 대신 RDD를 사용해야합니다. 당신의 날짜 열은 카산드라에 술어를 아래로 밀어 .where 기능을 사용할 수있는 클러스터링 키가

경우 : 카산드라에서 읽기에 대한 자세한 효율성 고체에 대한

val rdd = sc.cassandraTable("foo", "bar") 
val date = DateTime.now().minusDays(30) 
rdd.filter(r => r.getDate("date").after(date.toDate)) // Filters RDD 
println(rdd.cassandraCount()) // Ignores filtered rdd and counts everything 

---

. 그 외에는 데이터 서버 측을 정리할 수있는 것이 많지 않습니다.

https://github.com/datastax/spark-cassandra-connector/blob/master/doc/3_selection.md#filtering-rows---where