--- title: "Cloud Data Streaming Service로 Spark Streaming 연동" slug: "hadoop-vpc-use-ex13" tags: ["cloud hadoop", "Cloud Data Streaming Service"] updated: 2026-10-22T09:13:22Z published: 2026-10-22T09:13:22Z canonical: "guide-gov.ncloud-docs.com/hadoop-vpc-use-ex13" --- > ## Documentation Index > Fetch the complete documentation index at: https://guide-gov.ncloud-docs.com/llms.txt > Use this file to discover all available pages before exploring further. # Cloud Data Streaming Service로 Spark Streaming 연동 VPC 환경에서 이용 가능합니다. 이 가이드는 네이버 클라우드 플랫폼 Cloud Hadoop과 Cloud Data Streaming Service(CDSS)를 연동하는 방법을 소개합니다. ## 사전 작업 1. Cloud Data Streaming Service를 생성해 주십시오. - Cloud Data Streaming Service 생성에 관한 자세한 내용은 [Cloud Data Streaming Service 사용 가이드](/docs/cdss-overview)를 참고해 주십시오. 2. Cloud Data Streaming Service 사용하기 위해 VM 생성 및 세팅을 해주십시오. - VM 생성과 세팅에 관한 자세한 내용은 [Cloud Data Streaming Service 사용 가이드](/docs/cdss-overview)를 참고해 주십시오. 3. Cloud Hadoop 클러스터를 생성해 주십시오. - Cloud Hadoop 클러스터 생성에 관한 자세한 내용은 [Cloud Hadoop 시작](/docs/hadoop-chadoop-2-1-vpc) 가이드를 참고해 주십시오. 참고 Cloud Hadoop과 Cloud Data Streaming Service는 같은 VPC 내 통신이 가능한 동일 Subnet으로 생성하는 것을 권장합니다. 1. ACG를 설정해 주십시오. - Cloud Hadoop에서 Cloud Data Streaming Service Broker 노드에 접속하기 위해서 9092 포트를 허용해줘야 합니다. - Cloud Data Streaming Service의 **Broker 노드 ACG** 접근 소스에 Cloud Hadoop의 Subnet 대역을 추가해 주십시오. ![cloudhadoop-use-pre-vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/cloudhadoop-use-pre-vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D) 참고 Zeppelin Notebook에서 CDSS 연동하려면 추가적으로 Cloud Hadoop ACG에 9996 포트를 허용해주어야 합니다. 자세한 내용은 [Web UI 접속](/docs/chadoop-4-vpc#2-acg-%EB%A3%B0-%EC%84%A4%EC%A0%95) 가이드를 참고해 주십시오. ## Kafka를 활용한 데이터 전송 1. Cloud Data Streaming Service VM에서 Kafka를 실행해 주십시오. ``` [root@s17e27e0cf6c ~ ]# cd kafka_2.12-2.4.0 [root@s17e27e0cf6c kafka_2.12-2.4.0]# ./bin/kafka-server-start.sh -daemon config/server.properties ``` 1. 토픽을 생성해 주십시오. - bootstrap-server 뒤에는 broker-list를 넣어주십시오. ``` # 토픽 생성 [root@s17e27e0cf6c kafka_2.12-2.4.0]# ./bin/kafka-topics.sh --create --bootstrap-server 172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092 --replication-factor 1 --partitions 1 --topic [topic] # 생성된 토픽 확인 [root@s17e27e0cf6c kafka_2.12-2.4.0]# ./bin/kafka-topics.sh --list --bootstrap-server 172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092 ``` 참고 broker-list는 **Cloud Data Streaming Service > Cluster > Broker 노드 정보**에서 확인할 수 있습니다. 1. 데이터를 생성해 주십시오. ``` [root@s17e27e0cf6c kafka_2.12-2.4.0]# ./bin/kafka-console-producer.sh --broker-list 172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092 --topic [topic] ``` ## Kafka 연동 이 가이드에서는 Spark Streaming으로 Kafka 연동하는 두 가지 방법을 설명합니다. 1. 엣지 노드에서 CDSS 연동하기 2. Zeppelin Notebook에서 CDSS 연동하기 참고 Spark Streaming에 대한 자세한 내용은 [Spark Streaming Programming Guide](https://spark.apache.org/docs/latest/streaming-programming-guide.html)를 참고해 주십시오. ### 엣지 노드에서 CDSS 연동 1. Cloud Hadoop 엣지 노드에서 Spark를 실행해 주십시오. - 엣지 노드 접속에 대한 자세한 내용은 [SSH로 클러스터 접속](/docs/hadoop-chadoop-3-2-vpc#2-%ED%81%B4%EB%9F%AC%EC%8A%A4%ED%84%B0-%EB%85%B8%EB%93%9C-%EC%A0%91%EC%86%8D) 가이드를 참고해 주십시오. ``` # Cloud Hadoop 1.5 이상 버전 [sshuser@e-001-example-pzt-hd ~]$ sudo -u {계정명} spark-shell --master yarn --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.8 # Cloud Hadoop 1.4 버전 [sshuser@e-001-example-pzt-hd ~]$ sudo -u {계정명} spark-shell --master yarn --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.2 ``` 1. Spark Streaming을 사용하여 실시간으로 데이터를 읽어주십시오. ``` > import org.apache.spark.sql.streaming.Trigger > val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092").option("subscribe", "[topic]").option("startingOffsets", "earliest").load() > val stream = df.writeStream.trigger(Trigger.ProcessingTime("5 seconds")).outputMode("append").format("console").start().awaitTermination() ``` ![hadoop-vpc-use-ex13_edge1_vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/hadoop-vpc-use-ex13_edge1_vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D) 일반적인 batch로도 데이터를 읽어올 수 있습니다. ``` > val df = spark.read.format("kafka").option("kafka.bootstrap.servers", "172.16.0.6:9092,172.16.0.7:9092,172.16.0.8:9092").option("subscribe", "[topic]").option("startingOffsets","earliest").load() > df.show ``` 1. Spark Streaming을 사용하여 Kafka에 데이터를 작성할 수도 있습니다. Straming을 하기 전 엣지 노드에서 먼저 checkpoint를 만들어야 합니다. ``` hdfs dfs -mkdir -p /streaming/checkpointLocation ``` test 토픽에서 데이터를 읽어오고 읽어온 데이터를 새로운 토픽에 저장해 줍니다. ``` > val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092").option("subscribe", "test").option("startingOffsets", "earliest").load() > val ds = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)").writeStream.format("kafka").option("kafka.bootstrap.servers", "172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092").option("checkpointLocation","/streaming/checkpointLocation").option("topic", "newtopic").start() ``` 실시간으로 데이터를 처리할 필요가 없을 경우 아래 코드로 간단하게 내용을 저장할 수 있습니다. ``` > val df = spark.read.format("kafka").option("kafka.bootstrap.servers", "172.16.0.6:9092,172.16.0.7:9092,172.16.0.8:9092").option("subscribe", "test").option("startingOffsets","earliest").load() > df.selectExpr("key","value").write.format("kafka").option("kafka.bootstrap.servers","172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092").option("topic","newtopic").save() ``` 이제 Kafka에서 newtopic을 확인해 보면 데이터가 들어가 있는 걸 확인할 수 있습니다. ![hadoop-vpc-use-ex13_edge2_vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/hadoop-vpc-use-ex13_edge2_vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D) ### Zeppelin Notebook에서 CDSS 연동 1. Zeppelin UI에 접속한 후, Interpreter를 클릭해 주십시오. ![hadoop-vpc-use-ex13_zeppelin1_vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/hadoop-vpc-use-ex13_zeppelin1_vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D) 2. spark2 하위에 Dependencies를 추가해 주십시오. - artifact : org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.2 - exclude : net.jpountz.lz4:lz4:1.3.0 ![hadoop-vpc-use-ex13_zeppelin2_vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/hadoop-vpc-use-ex13_zeppelin2_vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D) 3. **[Notebook] > Create new note**를 클릭한 후, 새 노트북을 생성해 주십시오. - Default Interpreter는 spark2로 설정해 주십시오. ![hadoop-vpc-use-ex13_zeppelin3_vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/hadoop-vpc-use-ex13_zeppelin3_vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D) 4. Zeppelin Notebook에서 Spark Streaming을 사용하여 실시간으로 데이터를 읽고 쓸 수 있습니다. 코드는 다음과 같습니다. ``` > import org.apache.spark.sql.streaming.Trigger > val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "172.16.2.6:9092,172.16.2.7:9092,172.16.2.8:9092").option("subscribe", "[topic]").option("startingOffsets", "earliest").load() > val stream = df.writeStream.trigger(Trigger.ProcessingTime("5 seconds")).outputMode("append").format("console").start().awaitTermination() ``` ![hadoop-vpc-use-ex13_zeppelin4_vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/hadoop-vpc-use-ex13_zeppelin4_vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D) 참고 UDF를 이용해서 바이너리로 표현된 데이터를 string으로 변환할 수 있습니다. 아래는 예제코드입니다. ``` > import org.apache.spark.sql.functions.udf > val df = spark.read.format("kafka").option("kafka.bootstrap.servers", "172.16.0.6:9092,172.16.0.7:9092,172.16.0.8:9092").option("subscribe", "test").load() > val toString = udf((payload: Array[Byte]) => new String(payload)) > val chstring = df.withColumn("value", toStr(df("value"))) > df.show ``` ![hadoop-vpc-use-ex13_note1_vpc_ko](https://cdn.document360.io/f2d93c92-8957-408f-94df-b9b370f76a32/Images/Documentation/hadoop-vpc-use-ex13_note1_vpc_ko.png?sv=2026-02-06&spr=https&st=2026-10-29T11%3A49%3A01Z&se=2026-10-29T12%3A49%3A01Z&sr=c&sp=r&sig=ely95%2BGIQz22b12Yt4qcwBIZu%2BdtbpVAVuVyzSMAFs8%3D)