<div id="readability-page-1" class="page"><div>
												
<h2>플러그인 설치</h2>
<p>IntelliJ 에서 <code>Scala</code>, <code>Big Data Tools</code> 을 설치한다.</p>
<h2>Scala 버전 확인</h2>
<pre><code>spark-shell
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
22/04/16 09:45:23 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://ip-172-31-18-128.ap-northeast-2.compute.internal:4040
Spark context available as 'sc' (master = local[*], app id = local-1650102324143).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.2.1
      /_/

Using Scala version 2.12.15 (OpenJDK 64-Bit Server VM, Java 1.8.0_312)
Type in expressions to have them evaluated.
Type :help for more information.

scala&gt; :quit</code></pre>
<h2>프로젝트 생성하기</h2>
<p>Scala 를 선택하고, JDK 버전은 8 로 선택한다.</p>
<p><img src="https://www.skyer9.pe.kr/wordpress/wp-content/uploads/2022/04/2022-04-16-10.png" alt=""></p>
<h2>의존성 추가</h2>
<p>build.sbt 에 의존성을 추가해 준다.</p>
<pre><code>ThisBuild / version := "0.1.0-SNAPSHOT"

ThisBuild / scalaVersion := "2.12.15"

val sparkVersion = "3.2.1"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion,
  "org.apache.spark" %% "spark-sql" % sparkVersion
)

lazy val root = (project in file("."))
  .settings(
    name := "ReadTopic"
  )</code></pre>
<h2>Scala class 추가</h2>
<p>ReadTopic.scala 클래스를 추가해 준다.</p>
<pre><code>import org.apache.spark.sql.SparkSession

object ReadTopic {
  def main(args: Array[String]) {

    val kafkaBrokers = "test.skyer9.pe.kr:9092"
    val kafkaTopic = "exam";

    val spark = SparkSession.builder()
      .master("local[1]")
      .appName("Test")
      .getOrCreate();

    val df = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", kafkaBrokers)
      .option("subscribe", kafkaTopic)
      .option("startingOffsets", "earliest") // From starting
      .load()

    df.printSchema()
  }
}</code></pre>
<p>패키지를 인식하지 못하면 프로젝트를 닫았다 다시 열어준다.</p>
<h2>추가설정하기</h2>
<p><a href="https://www.skyer9.pe.kr/wordpress/?p=5252">여기</a> 를 참조해 설정해 준다.</p>
<p>프로젝트를 실행하면, jar 생성 후, 서버에 복사해 주고,<br>
Spark-Submit 까지 자동으로 해준다.</p>
<pre><code>/home/spark/spark-3.2.1-bin-hadoop3.2/bin/spark-submit --master local --deploy-mode client --class ReadTopic --name Remote --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 file://$HOME/ReadTopic.jar</code></pre>
<h2>프로젝트 실행하기</h2>
<pre><code>root
 |-- key: binary (nullable = true)
 |-- value: binary (nullable = true)
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- timestampType: integer (nullable = true)</code></pre>
<p><span></span>
				<span>Post Views:</span>
				<span>5</span>
			</p>											</div></div>

상구리의 기술 블로그

Spark 에서 Kafka Topic 읽어오기
플러그인 설치
IntelliJ 에서 Scala, Big Data Tools 을 설치한다.
Scala 버전 확인
spark-shell
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
22/04/16 09:45:23 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://ip-172-31-18-128.ap-northeast-2.compute.internal:4040
Spark context available as 'sc' (master = local[*], app id = local-1650102324143).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.2.1
      /_/
Using Scala version 2.12.15 (OpenJDK 64-Bit Server VM, Java 1.8.0_312)
Type in expressions to have them evaluated.
Type :help for more information.
scala> :quit
프로젝트 생성하기
Scala 를 선택하고, JDK 버전은 8 로 선택한다.
의존성 추가
build.sbt 에 의존성을 추가해 준다.
ThisBuild / version := "0.1.0-SNAPSHOT"
ThisBuild / scalaVersion := "2.12.15"
val sparkVersion = "3.2.1"
libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion,
  "org.apache.spark" %% "spark-sql" % sparkVersion
)
lazy val root = (project in file("."))
  .settings(
    name := "ReadTopic"
  )
Scala class 추가
ReadTopic.scala 클래스를 추가해 준다.
import org.apache.spark.sql.SparkSession
object ReadTopic {
  def main(args: Array[String]) {
    val kafkaBrokers = "test.skyer9.pe.kr:9092"
    val kafkaTopic = "exam";
    val spark = SparkSession.builder()
      .master("local[1]")
      .appName("Test")
      .getOrCreate();
    val df = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", kafkaBrokers)
      .option("subscribe", kafkaTopic)
      .option("startingOffsets", "earliest") // From starting
      .load()
    df.printSchema()
  }
}
패키지를 인식하지 못하면 프로젝트를 닫았다 다시 열어준다.
추가설정하기
여기 를 참조해 설정해 준다.
프로젝트를 실행하면, jar 생성 후, 서버에 복사해 주고,
/home/spark/spark-3.2.1-bin-hadoop3.2/bin/spark-submit --master local --deploy-mode client --class ReadTopic --name Remote --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 file://$HOME/ReadTopic.jar
프로젝트 실행하기
root
 |-- key: binary (nullable = true)
 |-- value: binary (nullable = true)
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- timestampType: integer (nullable = true)

Spark 에서 Kafka Topic 읽어오기

댓글