scala - 使用结构化流时无法将 json 格式数据写入路径。执行 spark2-submit 时仅创建 _spark_metadata
问题描述
我正在将 Kafka 结构化流中的 json 数据写入文件路径,当我从 shell 执行此操作时,我能够执行此操作。当我将它编译到一个 jar 并执行 spark2-submit 时,只创建了 _spark_metadata 并且没有找到数据
我尝试从 shell 执行此操作,并且能够在文件路径中看到 json 文件。我使用“sbt clean package”编译程序,然后尝试使用 spark-submit 运行它不会创建任何数据。
export SPARK_KAFKA_VERSION=0.10
spark2-submit --jars /opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/kafka-clients-0.9.0-kafka-2.0.2.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/spark-sql-kafka-0-10_2.11-2.3.0.cloudera5-20190107.080402-22.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/mongo-spark-connector_2.11-2.3.2.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/bson-3.9.0.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/mongo-java-driver-3.9.0.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/mongodb-driver-core-3.9.0.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/mongodb-driver-3.9.0.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/mongodb-driver-async-3.9.0.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/casbah-commons_2.11-2.8.2.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/casbah-core_2.11-2.8.2.jar,/opt/cloudera/parcels/CDH-5.13.3-1.cdh5.13.3.p0.2/jars/casbah-query_2.11-2.8.2.jar --class OSB_kafkaToSpark --master yarn --deploy-mode client /home/streaming_osb_2.11-0.1.0-SNAPSHOT.jar
import com.mongodb.client.MongoCollection
import com.mongodb.spark.config.WriteConfig
import com.mongodb.spark.{MongoConnector}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, _}
import scala.collection.JavaConverters._
import scala.collection.mutable
import org.apache.spark.sql._
import org.bson.Document
import org.apache.spark.sql.streaming.Trigger
object OSB_kafkaToSpark {
def main(args: Array[String]): Unit = {
val spark = SparkSession.
builder().
appName("OSB_kafkaToSpark").
getOrCreate()
println("SparkSession -> "+spark)
import spark.implicits._
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "bootstrap1.com:9092, bootstrap2.com:9092")
.option("subscribe", "topictest")
.option("failOnDataLoss", "false")
.load()
val dfs = df.selectExpr("CAST(value AS STRING)")
val data = dfs.withColumn("splitted", split($"value", "/"))
.select($"splitted".getItem(4).alias("region"), $"splitted".getItem(5).alias("service"), col("value"))
.withColumn("service_type", regexp_extract($"service", """.*(Inbound|Outbound|Outound).*""", 1))
.withColumn("region_type", concat(
when(col("region").isNotNull, col("region")).otherwise(lit("null")), lit(" "),
when(col("service").isNotNull, col("service_type")).otherwise(lit("null"))))
.withColumn("datetime", regexp_extract($"value", """\d{4}-[01]\d-[0-3]\d [0-2]\d:[0-5]\d:[0-5]\d""", 0))
val extractedDF = data.filter(
col("region").isNotNull &&
col("service").isNotNull &&
col("value").isNotNull &&
col("service_type").isNotNull &&
col("region_type").isNotNull &&
col("datetime").isNotNull)
.filter("region != ''")
.filter("service != ''")
.filter("value != ''")
.filter("service_type != ''")
.filter("region_type != ''")
.filter("datetime != ''")
val pathstring = "/user/spark_streaming".concat(args(0))
val query = extractedDF.writeStream
.format("json")
.option("path", pathstring)
.option("checkpointLocation", "/user/checkpoint")
.outputMode("append")
.trigger(Trigger.ProcessingTime("20 seconds"))
.start()
query.stop()
}
}
我想在运行 spark2-submit 时将流数据保存到一个文件夹中
解决方案
我想出了答案,我需要使用 query.awaitTermination()
推荐阅读
- gpu - 使用 CPU 工作人员进行数据处理,并使用带有 dask 的 GPU 工作人员训练 xgboost
- adfs - 使用 ADFS 作为 SAML IdP 的 Gigya - 如何将组添加到 Gigya 帐户
- mysql - 使用 hpa 扩展 mysql 在 Kubernetes 上创建一个空数据库
- ssis - SSIS DTSX 提取对象名称和 SQL 语句
- nginx - Nginx 重定向到不需要的端口
- sass - Netlify 部署
- macos - macOS Apple M1 上的 TensorFlow
- javascript - ReferenceError:客户端未定义 Discord.js
- superscript - 将 12 7-12 转换为上标 12⁷⁻¹/² 有没有办法减少 / 的长度,并且 12 7-1/2 在一个单元格中。我用下面的代码手动转换了它
- r - 为什么 R 在 0.3 的值上挣扎?