{"id":5270,"date":"2022-04-16T17:52:27","date_gmt":"2022-04-16T08:52:27","guid":{"rendered":"https:\/\/www.skyer9.pe.kr\/wordpress\/?p=5270"},"modified":"2022-04-19T15:17:17","modified_gmt":"2022-04-19T06:17:17","slug":"spark-%ec%97%90%ec%84%9c-kafka-topic-%ec%9d%bd%ec%96%b4%ec%98%a4%ea%b8%b0","status":"publish","type":"post","link":"https:\/\/www.skyer9.pe.kr\/wordpress\/?p=5270","title":{"rendered":"Spark \uc5d0\uc11c Kafka Topic \uc77d\uc5b4\uc624\uae30"},"content":{"rendered":"

Spark \uc5d0\uc11c Kafka Topic \uc77d\uc5b4\uc624\uae30<\/h1>\n

\ud50c\ub7ec\uadf8\uc778 \uc124\uce58<\/h2>\n

IntelliJ \uc5d0\uc11c Scala<\/code>, Big Data Tools<\/code> \uc744 \uc124\uce58\ud55c\ub2e4.<\/p>\n

Scala \ubc84\uc804 \ud655\uc778<\/h2>\n
spark-shell\nUsing Spark's default log4j profile: org\/apache\/spark\/log4j-defaults.properties\nSetting default log level to "WARN".\nTo adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n22\/04\/16 09:45:23 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\nSpark context Web UI available at http:\/\/ip-172-31-18-128.ap-northeast-2.compute.internal:4040\nSpark context available as 'sc' (master = local[*], app id = local-1650102324143).\nSpark session available as 'spark'.\nWelcome to\n      ____              __\n     \/ __\/__  ___ _____\/ \/__\n    _\\ \\\/ _ \\\/ _ `\/ __\/  '_\/\n   \/___\/ .__\/\\_,_\/_\/ \/_\/\\_\\   version 3.2.1\n      \/_\/\n\nUsing Scala version 2.12.15 (OpenJDK 64-Bit Server VM, Java 1.8.0_312)\nType in expressions to have them evaluated.\nType :help for more information.\n\nscala> :quit<\/code><\/pre>\n
\ud504\ub85c\uc81d\ud2b8 \uc0dd\uc131\ud558\uae30<\/h2>\n
Scala \ub97c \uc120\ud0dd\ud558\uace0, JDK \ubc84\uc804\uc740 8 \ub85c \uc120\ud0dd\ud55c\ub2e4.
\n\ud655\uc778\ud55c Scala \ubc84\uc804\uc744 \uc124\uc815\ud574 \uc900\ub2e4.<\/p>\n
\"\"<\/p>\n
\ucd94\uac00\uc124\uc815\ud558\uae30<\/h2>\n
\uc5ec\uae30<\/a> \ub97c \ucc38\uc870\ud574 \uc124\uc815\ud574 \uc900\ub2e4.<\/p>\n
\ud504\ub85c\uc81d\ud2b8\ub97c \uc2e4\ud589\ud558\uba74, jar \uc0dd\uc131 \ud6c4, \uc11c\ubc84\uc5d0 \ubcf5\uc0ac\ud574 \uc8fc\uace0,
\nSpark-Submit \uae4c\uc9c0 \uc790\ub3d9\uc73c\ub85c \ud574\uc900\ub2e4.<\/p>\n
\uc6b0\uc120\uc740 \uc815\uc0c1\uc2e4\ud589\ub9cc \ud655\uc778\ud574 \ubcf8\ub2e4.<\/p>\n
\/home\/spark\/spark-3.2.1-bin-hadoop3.2\/bin\/spark-submit --master local --deploy-mode client --name Unnamed file:\/\/$HOME\/ReadTopic.jar<\/code><\/pre>\n
\uc758\uc874\uc131 \ucd94\uac00<\/h2>\n
build.sbt \uc5d0 \uc758\uc874\uc131\uc744 \ucd94\uac00\ud574 \uc900\ub2e4.<\/p>\n
ThisBuild \/ version := "0.1.0-SNAPSHOT"\n\nThisBuild \/ scalaVersion := "2.12.15"\n\nval sparkVersion = "3.2.1"\n\nlibraryDependencies ++= Seq(\n  "org.apache.spark" %% "spark-core" % sparkVersion,\n  "org.apache.spark" %% "spark-sql" % sparkVersion\n)\n\nlazy val root = (project in file("."))\n  .settings(\n    name := "ReadTopic"\n  )<\/code><\/pre>\n
Scala class \uc218\uc815<\/h2>\n
Main.scala \ud074\ub798\uc2a4\ub97c \uc218\uc815\ud574 \uc900\ub2e4.<\/p>\n
import org.apache.spark.sql.SparkSession\n\nobject ReadTopic {\n  def main(args: Array[String]) {\n\n    val kafkaBrokers = "test.skyer9.pe.kr:9092"\n    val kafkaTopic = "exam"\n\n    val spark = SparkSession.builder()\n      .master("local[1]")\n      .appName("Test")\n      .getOrCreate()\n\n    val df = spark.readStream\n      .format("kafka")\n      .option("kafka.bootstrap.servers", kafkaBrokers)\n      .option("subscribe", kafkaTopic)\n      .option("startingOffsets", "earliest") \/\/ From starting\n      .load()\n\n    df.printSchema()\n  }\n}<\/code><\/pre>\n
\ud328\ud0a4\uc9c0\ub97c \uc778\uc2dd\ud558\uc9c0 \ubabb\ud558\uba74 \ud504\ub85c\uc81d\ud2b8\ub97c \ub2eb\uc558\ub2e4 \ub2e4\uc2dc \uc5f4\uc5b4\uc900\ub2e4.<\/p>\n
\ucd94\uac00\uc124\uc815\ud558\uae30<\/h2>\n
\ud328\ud0a4\uc9c0 org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 \uac00 \ucd94\uac00\ub418\uc5b4\uc57c \ud55c\ub2e4.<\/p>\n
\/home\/spark\/spark-3.2.1-bin-hadoop3.2\/bin\/spark-submit --master local --deploy-mode client --name Remote --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 file:\/\/$HOME\/ReadTopic.jar<\/code><\/pre>\n
\ud504\ub85c\uc81d\ud2b8 \uc2e4\ud589\ud558\uae30<\/h2>\n
key-value \ud615\uc2dd\uc73c\ub85c \ub370\uc774\ud0c0\uac00 \uc218\uc2e0\ub41c\ub2e4.<\/p>\n
root\n |-- key: binary (nullable = true)\n |-- value: binary (nullable = true)\n |-- topic: string (nullable = true)\n |-- partition: integer (nullable = true)\n |-- offset: long (nullable = true)\n |-- timestamp: timestamp (nullable = true)\n |-- timestampType: integer (nullable = true)<\/code><\/pre>\n
\ud654\uba74\uc5d0 \ucd9c\ub825\ud558\uae30<\/h2>\n
IntelliJ \ucf58\uc194\uc5d0\uc11c \ud55c\uae00\uc774 \uae68\uc9c0\ub294 \uac83\uc740 \uc2e0\uacbd\uc4f8 \ud544\uc694\uc5c6\ub2e4.
\n(IntelliJ \ubb38\uc81c)
\n\uc11c\ubc84\uc5d0\uc11c \uc9c1\uc811 \uc2e4\ud589\ud574 \ubcf4\uba74 \uc815\uc0c1\uc801\uc73c\ub85c \ucd9c\ub825\ub41c\ub2e4.<\/p>\n
    val kafkaBrokers = "test.skyer9.pe.kr:9092"\n    val kafkaTopic = "json_topic"\n\n    val spark = SparkSession.builder()\n      .master("local[1]")\n      .appName("Test")\n      .getOrCreate()\n\n    val df = spark.readStream\n      .format("kafka")\n      .option("kafka.bootstrap.servers", kafkaBrokers)\n      .option("subscribe", kafkaTopic)\n      .option("startingOffsets", "earliest") \/\/ From starting\n      .load()\n\n    val jsonStringDF = df.selectExpr("CAST(value AS STRING)")\n\n    val schema = new StructType()\n      .add("dt",StringType)\n      .add("msg",StringType)\n\n    val jsonDF = jsonStringDF.select(from_json(col("value"), schema).as("data"))\n      .select("data.*")\n\n    jsonDF.writeStream\n      .format("console")\n      .outputMode("append")\n      .start()\n      .awaitTermination()     \/\/ \uc785\ub825\uc744 \uacc4\uc18d \uae30\ub2e4\ub9bc<\/code><\/pre>\n
hadoop \uc5d0 \uc800\uc7a5\ud558\uae30<\/h2>\n
hadoop fs -rm -r \/streaming\/checkpointLocation\nhadoop fs -rm -r \/streaming\/out\/*\n\nhadoop fs -mkdir -p \/tmp\nhadoop fs -chmod 777 \/tmp\n\nhadoop fs -mkdir -p \/streaming\/out\nhadoop fs -mkdir -p \/streaming\/checkpointLocation<\/code><\/pre>\n
    jsonDF.writeStream\n      .format("com.databricks.spark.csv")\n      .outputMode("append")\n      .option("path", "hdfs:\/\/localhost:9000\/streaming\/out")\n      .option("checkpointLocation", "hdfs:\/\/localhost:9000\/streaming\/checkpointLocation")\n      .start()\n      .awaitTermination()<\/code><\/pre>\n
hadoop \uc5d0 parquet(\ud30c\ucf00\uc774) \ud3ec\uba67\uc73c\ub85c \uc800\uc7a5\ud558\uae30<\/h2>\n
hadoop fs -rm -r \/streaming\/checkpointLocation\nhadoop fs -rm -r \/streaming\/out\/*\n\nhadoop fs -mkdir -p \/streaming\/out\nhadoop fs -mkdir -p \/streaming\/checkpointLocation<\/code><\/pre>\n
    jsonDF.writeStream\n      .format("parquet")\n      .outputMode("append")\n      .option("path", "hdfs:\/\/localhost:9000\/streaming\/out")\n      .option("checkpointLocation", "hdfs:\/\/localhost:9000\/streaming\/checkpointLocation")\n      .start()\n      .awaitTermination()<\/code><\/pre>\n
hive \uc5d0\uc11c \ub370\uc774\ud0c0 \uc77d\uae30<\/h2>\n
hive\n\nhive (default)> CREATE EXTERNAL TABLE tbl_weblog (\ndt string,\nmsg string\n)\nSTORED AS PARQUET\nLOCATION "hdfs:\/\/localhost:9000\/streaming\/out";\n\nhive (default)> select * from tbl_weblog;\n\nhive (default)> select * from tbl_weblog order by dt desc limit 10;<\/code><\/pre>\n","protected":false},"excerpt":{"rendered":"
Spark \uc5d0\uc11c Kafka Topic \uc77d\uc5b4\uc624\uae30 \ud50c\ub7ec\uadf8\uc778 \uc124\uce58 IntelliJ \uc5d0\uc11c Scala, Big Data Tools \uc744 \uc124\uce58\ud55c\ub2e4. Scala \ubc84\uc804 \ud655\uc778 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\u2026 Read More »<\/a><\/span><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[10],"tags":[],"class_list":["post-5270","post","type-post","status-publish","format-standard","hentry","category-kafka"],"_links":{"self":[{"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/5270","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=5270"}],"version-history":[{"count":19,"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/5270\/revisions"}],"predecessor-version":[{"id":5309,"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/5270\/revisions\/5309"}],"wp:attachment":[{"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=5270"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=5270"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.skyer9.pe.kr\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=5270"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}