SnappyData + Zeppelin + Kafka 스트리밍 - 스트리밍 테이블을 만드는 중 오류가 발생했습니다.

Question

Zeppelin을 사용하여 SnappyData 스트리밍 테이블을 만들려고합니다. 나는 몇 단락으로 나누어 져 'rowConverter'

제플린 노트북 인수에 스트림 테이블 정의와 문제가 :

조 제 1 항 :

import org.apache.spark.sql.Row 
import org.apache.spark.sql.streaming.{SchemaDStream, StreamToRowsConverter} 

class RowsConverter extends StreamToRowsConverter with Serializable { 

override def toRows(message: Any): Seq[Row] = { 
val log = message.asInstanceOf[String] 
val fields = log.split(",") 
val rows = Seq(Row.fromSeq(Seq(new java.sql.Timestamp(fields(0).toLong), 
    fields(1), 
    fields(2), 
    fields(3), 
    fields(4), 
    fields(5).toDouble, 
    fields(6) 
))) 

rows 
} 
} 

제 2 항 :

snsc.sql(
"CREATE STREAM TABLE adImpressionStream if not exists ("sensor_id string, metric 
metric string) using kafka_stream 
options (storagelevel 'MEMORY_AND_DISK_SER_2', 
rowConverter 'RowsConverter', 
zkQuorum 'localhost:2181', 
groupId 'streamConsumer', topics 'test'");" 
) 

첫 번째 단락에서 오류를 반환합니다.

(210)

두 번째 문단 :

java.lang.RuntimeException: Failed to load class : java.lang.ClassNotFoundException: RowsConverter 

내가 자식에서 기본 코드를 사용하는 것을 시도하고있다 :

snsc.sql("create stream table streamTable (userId string, clickStreamLog string) " + 
"using kafka_stream options (" + 
"storagelevel 'MEMORY_AND_DISK_SER_2', " + 
" rowConverter 'io.snappydata.app.streaming.KafkaStreamToRowsConverter' ," + 
"kafkaParams 'zookeeper.connect->localhost:2181;auto.offset.reset->smallest;group.id->myGroupId', " + 
"topics 'test')") 

하지만 비슷한 오류가 있습니다

java.lang.RuntimeException: Failed to load class : java.lang.ClassNotFoundException: io.snappydata.app.streaming.KafkaStreamToRowsConverter 

당신이 좀 도와 수 이 문제? 정말 고마워요.

Answer 1

응용 프로그램 특정 클래스를 클래스 경로에 제공해야합니다. 여기 클래스 패스 설정 단계를 참조하십시오. 펠린에서 클래스 경로 설정을 선택할 것이다 spark-env.sh https://github.com/SnappyDataInc/snappy-poc#lets-get-this-going

Answer 2

여기에 주어진 아파치 제플린에 snappydata 인터프리터를 추가 https://snappydatainc.github.io/snappydata/howto/use_apache_zeppelin_with_snappydata/

이이 이기고 제플린을 실행 가능하게 할 것이다 코드가 실행되도록 임베디드 모드. 특히 클러스터 구성에서 "-classpath"옵션을 사용하여 필요한 jar를 설정해야합니다.