问题描述
我使用的是 Spark Structure Streaming,代码如下:
I am using Spark Structure Streaming, the code is in following:
def convert_timestamp_to_datetime(timestamp):
return datetime.fromtimestamp(timestamp)
def extract():
spark = SparkSession \
.builder \
.appName("StructuredNetworkWordCount") \
.getOrCreate()
json_schema = \
StructType() \
.add(StructField("TIMESTAMP", FloatType(), True)) \
.add(StructField("index", IntegerType(), True)) \
.add(StructField("CUSTOMER_ID", StringType(), True)) \
.add(StructField("CODE_ID", StringType(), True)) \
.add(StructField("PROCESS", StringType(), True))
my_df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "localhost:9092") \
.option("subscribe", "simple_json_12_10trx") \
.option("startingOffsets", "earliest") \
.load()
my_df = my_df.select(from_json(col('value').cast('string'), json_schema).alias("json"))
convert_timestamp_datetime_udf = udf(lambda x: convert_timestamp_to_datetime(x), TimestampType())
return my_df.select('json.*', convert_timestamp_datetime_udf('json.TIMESTAMP').alias('DATETIME'))
def transform_load(my_df, epoch_id):
update_obj = my_df.groupBy('CUSTOMER_ID').agg(F.count('CUSTOMER_ID').alias('count_t'),F.collect_set('CODE_ID').alias('unique_CODE'))
update_obj.show()
update(update_obj)
if __name__ == '__main__':
start = time.time()
df = extract()
query = df.writeStream \
.outputMode('append')\
.foreachBatch(transform_load)\
.start() \
.awaitTermination()
我想访问分布式的每一行 Spark Dataframe.所以,我必须使用 Map 转换.我只是添加了这个简单的代码来测试 Spark Map.但是,我在控制台中没有收到任何输出.实际上,func
并没有运行.
I want to access each row of Spark Dataframe distributed. So, I must use Map transformation. I just add this simple code to test Spark Map. But, I do not receive any output in console. In fact, func
is not run.
def func(df):
df.take(3)
def update(df):
df.rdd.map(func,preservesPartitioning=False)
你能指导我这里有什么问题吗?
Would you please guide me what is wrong here?
非常感谢.
推荐答案
问题已解决.
我忘记在 map 之后使用 action
.另外,我不能使用 df.take(3)
因为 func
中没有任何数据帧,它是 rdd
没有 take
属性.我像这样更改代码:
I forgot to use action
after map. Also, I cannot use df.take(3)
since there is not any dataframe in func
, it is rdd
which does not have take
attribute. I change the code like this :
def func(x):
print(x.CUSTOMER_ID)
def update(df):
df.rdd.map(func,preservesPartitioning=False).count()
count()
是我用来查看地图结果的操作.
count()
is an action that I use to see map result.
这篇关于如何在 Python 中的 Spark Dataframe 上应用任何类型的地图转换的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!