Spark를 사용하여 카산드라에서 카우치베이스로 데이터 전송하기

시반쉬 스리바스타바 는 다국어 개발자이자 스칼라, 스파크 애호가입니다. 그는 오픈소스 프로젝트에 기여하는 것을 좋아하며 Apache Iota, Apache Spark, Apache Carbondata와 같은 많은 프로젝트에 기여했습니다, 카우치베이스-스파크-커넥터, Akka, Alpakka 및 기타 여러 프로젝트에 참여했습니다. 그는 IoT, 딥러닝 등과 같은 향후 기술에 큰 관심을 가지고 있습니다. C현재 IoT 기반 스타트업인 Chirpanywhere Inc에서 수석 소프트웨어 엔지니어로 근무하며 하드웨어 프로그래밍부터 전체 솔루션 설계 및 배포에 이르기까지 다양한 업무를 관리하고 있습니다. 

많은 NoSQL 데이터베이스 카산드라, 몽고DB, 카우치베이스 등 다양한 솔루션이 시장에 나와 있으며, 각각 장단점이 있습니다.

NoSQL 데이터베이스의 유형

NoSQL 데이터베이스에는 크게 네 가지 유형이 있습니다:

1. 열 중심

2. 키-값 저장소

3. 문서 중심

4. 그래프

키-값 및 문서 지향 데이터베이스를 지원하는 Couchbase와 같이 두 가지 이상의 형식을 지원하는 데이터베이스를 '다중 모델'이라고 합니다.

애플리케이션에 맞는 데이터베이스를 잘못 선택했다가 나중에 이 냉혹한 진실을 깨닫는 경우가 있습니다.

그럼 어떻게 해야 하나요? 어떻게 해야 할까요?

저희의 경험도 마찬가지였는데, Cassandra를 데이터베이스로 사용하다가 나중에 이 데이터베이스가 저희의 모든 요구를 충족시키지 못한다는 사실을 알게 되었습니다. 새로운 데이터베이스를 찾던 중 Couchbase가 가장 적합하다는 것을 알게 되었습니다.

가장 큰 어려움은 다음에서 데이터를 어떻게 전송해야 하는지 파악하는 것이었습니다. 카산드라에서 카우치베이스로플러그인을 사용할 수 없었기 때문입니다.

이 블로그 게시물에서는 Spark를 사용하여 Cassandra에서 Couchbase로 데이터를 전송하는 코드를 설명하겠습니다.

모든 코드는 여기에서 확인할 수 있습니다: 카산드라-카우치베이스-전송-플러그인

코드 설명

여기서는 Cassandra에서 데이터를 읽고 Couchbase에 다시 쓰고 있습니다. 이 간단한 코드로 문제가 해결됩니다.

관련 단계는 다음과 같습니다:

1. 구성 읽기:
   val 구성 = ConfigFactory.load()
   //Couchbase 구성
   val 버킷 이름 = 구성.getString("couchbase.bucketName")
   val 카우치베이스 호스트 = 구성.getString("couchbase.host")
   //Cassandra 구성
   val 키공간 이름 = 구성.getString("cassandra.keyspaceName")
   val 테이블 이름 = 구성.getString("cassandra.tableName")
   val idFeild = 구성.getString("cassandra.idFeild")
   val 카산드라 호스트 = 구성.getString("cassandra.host")
   val 카산드라포트 = 구성.getInt("cassandra.port")

2. 스파크 구성 설정 및 스파크 세션 만들기:
   val conf = new SparkConf()
     .setAppName(s"Couchbase카산드라전송플러그인")
    .setMaster("local[*]")
    .set(s"com.couchbase.bucket.$버킷 이름“, “”)
    .set("com.couchbase.nodes", 카우치베이스 호스트)
    .set("spark.cassandra.connection.host", 카산드라 호스트)
    .set("spark.cassandra.connection.port", 카산드라포트.toString)
   val 스파크 = SparkSession.빌더().config(conf).getOrCreate()
   val sc = 스파크.sparkContext

3. 카산드라에서 데이터를 읽습니다:
     val cassandraRDD = 스파크.read
     .format("org.apache.spark.sql.cassandra")
     .options(지도("table" -> 테이블 이름, "키스페이스" -> 키공간 이름))
     .load()

4. ID 필드를 확인합니다:
   ID 필드가 존재하는지 확인 중입니다. 그런 다음 Couchbase에서도 이를 아이디로 사용하거나 임의의 아이디를 생성하여 문서에 할당합니다.
   가져오기 org.apache.spark.sql.functions._
   val uuidUDF = udf(카우치베이스헬퍼.getUUID _)
   val rddToBeWritten = 만약 (cassandraRDD.columns.contains(idFeild)) {
     cassandraRDD.withColumn("META_ID", cassandraRDD(idFeild))
   } else {
     cassandraRDD.withColumn("META_ID", uuidUDF())
   }
   
   다른 파일에 저장합니다:
   객체 카우치베이스 헬퍼 {
   def getUUID: 문자열 = UUID.randomUUID().toString
   }

5. 카우치베이스에 쓰기:
   rddToBeWritten.write.couchbase()

이 코드를 직접 실행하여 Cassandra에서 Couchbase로 데이터를 전송할 수 있습니다. - 몇 가지 설정만 하면 됩니다.

구성

모든 구성은 다음을 설정하여 수행할 수 있습니다. 환경 변수.

1. 카우치베이스 구성:

2. 카산드라 구성:

실행 중인 코드

카산드라 쪽:

이것이 Cassandra 측에서 데이터가 보이는 방식입니다.

카우치베이스 쪽:

사례 1: 아이디가 존재하고 Couchbase 문서 아이디와 동일하게 사용할 수 있는 경우.

사례 2: 아이디 이름이 존재하지 않아 문서에 임의의 아이디를 할당해야 하는 경우.

카산드라-카우치베이스 전송 플러그인을 실행하는 방법

코드를 실행하는 단계:

1. 리포지토리에서 코드를 다운로드하세요.

2. 구성에 따라 환경 변수를 구성합니다.

3. 다음을 사용하여 프로젝트를 실행합니다. SBT 실행

카우치베이스에 대해 글을 쓰고 커뮤니티 글쓰기 프로그램에 참여하고 싶으신가요? 자세히 알아보세요!