如何解决添加新行时,如何将一个csv文件的writeStream写入控制台作为接收器?
我试图每5分钟读取一次csv文件,并将少于5分钟前添加的所有新行输出到另一个csv文件。我认为,当我手动将新的一行数据添加到csv文件中时,一个不错的垫脚石是readStream一个csv文件和writeStream到控制台以打印出文件内容。问题是只有在我将一个全新的csv文件插入目录时,控制台才会打印。
每当将新行添加到csv文件时如何打印?
这是我的目录布局:
- stream_script2.py(下面的代码)
- csv_files
- unit_testing.csv
from pyspark.sql import SparkSession
from pyspark.sql.types import (
StringType,IntegerType,StructType,StructField,TimestampType,BooleanType,DateType,)
# Create SparkSession
spark = (
SparkSession.builder.appName("streaming")
.master("local[*]")
.getOrCreate()
)
# Made Schema
schema = StructType(
[
StructField("Drug_Name",StringType(),True),StructField("Count",IntegerType(),StructField("Faulty",BooleanType(),]
)
# Read Stream
df = (
spark.readStream.option("sep",";")
.schema(schema)
.format("csv")
.load("csv_files/unit_testing*.csv")
)
# Write Stream
query = df.writeStream
.format("console")
.outputMode("append")
.queryName("test1")
.start()
query.awaitTermination()
# Ctrl-C to stop stream
解决方法
我几乎可以肯定,Spark结构化流不直接支持在添加新行时通知我的功能。相反,解决方法是以某种方式保留对旧文件的引用,并将行数与当前文件进行比较。如果存在差异,则必须进行比较和提取。否则什么也不做。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。