请问为什么不能够在DF的foreachPartition方法调用的函数方法中引用redis模块呢？
我想请问下各位大佬。 # 为什么不能够在Dataframe的foreachPartition()的方法调用的函数方法中引用redis模块呢？ # 应该怎样解决这个问题呢？ #我使用的编程环境是jupyter。具体的操作场景为希望能够通过使用dataframe的foreachPartition方法和自定义的函数将用户DF和广告DF 的信息录入到redis数据库中去。
下面是我的代码：
feature_cols_from_ad=[
"price"
]
feature_cols_from_user=[
"cms_groupid_id",
"final_gender_code",
"age_level",
"shopping_level",
"occupation",
"pvalue_level",
"new_user_class_level"
]
def foreachPartition(partition):
import redis
import json
client01=redis.StrictRedis(host="192.168.56.5",port=6379,db=10)
for r in partition:
data={
"price":r.price
}
client01.hset("ad_features",r.adgroudId,json.dumps(data))
def foreachPartition2(partition):
import redis
import json
client02=redis.StrictRedis(host="192.168.56.5",port=6379,db=10)
for r in partition:
data={
"cms_group_id":r.cms_group_id,
"final_gender_code":r.final_gender_code,
"age_level":r.age_level,
"shopping_level":r.shopping_level,
"occupation":r.occupation,
"pvalue_level":r.pvalue_level,
"new_user_class_level":r.new_user_class_level
}
client02.hset("user_features",r.userId,json.dumps(data))
ad_feature_df.foreachPartition(foreachPartition)
user_profile_df.foreachPartition(foreachPartition2)
下面是具体的报错信息：
22/06/02 21:37:52 WARN TaskSetManager: Lost task 1.0 in stage 376.0 (TID 1054) (192.168.56.6 executor 0): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/sft/spark/python/lib/pyspark.zip/pyspark/worker.py", line 619, in main
process()
File "/usr/sft/spark/python/lib/pyspark.zip/pyspark/worker.py", line 609, in process
out_iter = func(split_index, iterator)
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 417, in func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 933, in func
File "/tmp/ipykernel_2896/4051490123.py", line 17, in foreachPartition
ModuleNotFoundError: No module named 'redis'
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:555)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:713)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:695)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:508)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
at scala.collection.generic.Growable.$plus$plus$eq(Growable.scala:62)
at scala.collection.generic.Growable.$plus$plus$eq$(Growable.scala:53)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:105)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:49)
at scala.collection.TraversableOnce.to(TraversableOnce.scala:366)
at scala.collection.TraversableOnce.to$(TraversableOnce.scala:364)
at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce.toBuffer(TraversableOnce.scala:358)
at scala.collection.TraversableOnce.toBuffer$(TraversableOnce.scala:358)
at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce.toArray(TraversableOnce.scala:345)
at scala.collection.TraversableOnce.toArray$(TraversableOnce.scala:339)
at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
at org.apache.spark.rdd.RDD.$anonfun$collect$2(RDD.scala:1030)
at org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2254)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
22/06/02 21:37:52 ERROR TaskSetManager: Task 0 in stage 376.0 failed 4 times; aborting job
22/06/02 21:37:52 WARN TaskSetManager: Lost task 1.3 in stage 376.0 (TID 1060) (192.168.56.6 executor 0): TaskKilled (Stage cancelled)
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
/tmp/ipykernel_2896/2204389708.py in <module>
----> 1 ad_feature_df.foreachPartition(foreachPartition)
2 user_profile_df.foreachPartition(foreachPartition2)
/usr/local/python3/lib/python3.7/site-packages/pyspark/sql/dataframe.py in foreachPartition(self, f)
791 >>> df.foreachPartition(f)
792 """
--> 793 self.rdd.foreachPartition(f)
794
795 def cache(self):
/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py in foreachPartition(self, f)
936 except TypeError:
937 return iter([])
--> 938 self.mapPartitions(func).count() # Force evaluation
939
940 def collect(self):
/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py in count(self)
1235 3
1236 """
-> 1237 return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
1238
1239 def stats(self):
/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py in sum(self)
1224 6.0
1225 """
-> 1226 return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)
1227
1228 def count(self):
/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py in fold(self, zeroValue, op)
1078 # zeroValue provided to each partition is unique from the one provided
1079 # to the final reduce call
-> 1080 vals = self.mapPartitions(func).collect()
1081 return reduce(op, vals, zeroValue)
1082
/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py in collect(self)
948 """
949 with SCCallSiteSync(self.context) as css:
--> 950 sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
951 return list(_load_from_socket(sock_info, self._jrdd_deserializer))
952
/usr/local/python3/lib/python3.7/site-packages/py4j/java_gateway.py in __call__(self, *args)
1320 answer = self.gateway_client.send_command(command)
1321 return_value = get_return_value(
-> 1322 answer, self.gateway_client, self.target_id, self.name)
1323
1324 for temp_arg in temp_args:
/usr/local/python3/lib/python3.7/site-packages/pyspark/sql/utils.py in deco(*a, **kw)
109 def deco(*a, **kw):
110 try:
--> 111 return f(*a, **kw)
112 except py4j.protocol.Py4JJavaError as e:
113 converted = convert_exception(e.java_exception)
/usr/local/python3/lib/python3.7/site-packages/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 376.0 failed 4 times, most recent failure: Lost task 0.3 in stage 376.0 (TID 1059) (192.168.56.6 executor 0): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/sft/spark/python/lib/pyspark.zip/pyspark/worker.py", line 619, in main
process()
File "/usr/sft/spark/python/lib/pyspark.zip/pyspark/worker.py", line 609, in process
out_iter = func(split_index, iterator)
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 417, in func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 933, in func
File "/tmp/ipykernel_2896/4051490123.py", line 17, in foreachPartition
ModuleNotFoundError: No module named 'redis'
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:555)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:713)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:695)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:508)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
at scala.collection.generic.Growable.$plus$plus$eq(Growable.scala:62)
at scala.collection.generic.Growable.$plus$plus$eq$(Growable.scala:53)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:105)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:49)
at scala.collection.TraversableOnce.to(TraversableOnce.scala:366)
at scala.collection.TraversableOnce.to$(TraversableOnce.scala:364)
at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce.toBuffer(TraversableOnce.scala:358)
at scala.collection.TraversableOnce.toBuffer$(TraversableOnce.scala:358)
at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce.toArray(TraversableOnce.scala:345)
at scala.collection.TraversableOnce.toArray$(TraversableOnce.scala:339)
at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
at org.apache.spark.rdd.RDD.$anonfun$collect$2(RDD.scala:1030)
at org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2254)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2454)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2403)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2402)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2402)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1160)
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1160)
at scala.Option.foreach(Option.scala:407)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1160)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2642)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2584)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2573)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:938)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2214)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2235)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2254)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2279)
at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1030)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:414)
at org.apache.spark.rdd.RDD.collect(RDD.scala:1029)
at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:180)
at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
at sun.reflect.GeneratedMethodAccessor184.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/sft/spark/python/lib/pyspark.zip/pyspark/worker.py", line 619, in main
process()
File "/usr/sft/spark/python/lib/pyspark.zip/pyspark/worker.py", line 609, in process
out_iter = func(split_index, iterator)
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 2918, in pipeline_func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 417, in func
File "/usr/local/python3/lib/python3.7/site-packages/pyspark/rdd.py", line 933, in func
File "/tmp/ipykernel_2896/4051490123.py", line 17, in foreachPartition
ModuleNotFoundError: No module named 'redis'
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:555)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:713)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:695)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:508)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
at scala.collection.generic.Growable.$plus$plus$eq(Growable.scala:62)
at scala.collection.generic.Growable.$plus$plus$eq$(Growable.scala:53)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:105)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:49)
at scala.collection.TraversableOnce.to(TraversableOnce.scala:366)
at scala.collection.TraversableOnce.to$(TraversableOnce.scala:364)
at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce.toBuffer(TraversableOnce.scala:358)
at scala.collection.TraversableOnce.toBuffer$(TraversableOnce.scala:358)
at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce.toArray(TraversableOnce.scala:345)
at scala.collection.TraversableOnce.toArray$(TraversableOnce.scala:339)
at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
at org.apache.spark.rdd.RDD.$anonfun$collect$2(RDD.scala:1030)
at org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2254)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
展开
收起
问答地址：
开发者社区
>
大数据与机器学习
>
开源大数据平台 E-MapReduce
>
问答
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
相关问答
阿里云EMR是云原生开源大数据平台，为客户提供简单易集成的Hadoop、Hive、Spark、Flink、Presto、ClickHouse、StarRocks、Delta、Hudi等开源大数据计算和存储引擎，计算资源可以根据业务的需要调整。EMR可以部署在阿里云公有云的ECS和ACK平台。
热门讨论
热门文章
hdfs 文件块过多（1亿）每个datanode 占3000万块(分16G内),某时刻内存陡然增加？
594
StarRocks有字符集的概念吗？有utf8和utf8mb4吗？有字符集的概念吗？有utf8和ut
1114
starRock 一定要 达到 8核，16核的要求 才能部署吗？4核 64G内存，行不行？
519
请问flink sql 写oss有个报错该怎么办？
1500
想问下，maxcomputer 创建的SQL脚本 可以用shell 调用运行吗
1133
Spark on k8s的核心概念是什么？
802
数据湖存储OSS是什么？
2536
cdh（6.3.2版本）集群访问oss存储的配置步骤
8438
乐观并发控制和悲观并发控制是区别是什么？
404
我想问一下阿里云E-MapReduce，为什么会报错这个信息呢？
217
展开全部
都是default惹的祸-yarn调度（一）-fair调度器drf调度策略作业不执行问题的调查和源码分析
4052
Apache Spark Meetup China 第1期 最全资料下载
2306
Delta元数据解析
2086
Data Lake 三剑客——Delta、Hudi、Iceberg 对比分析
15969
阿里巴巴的通义千问大模型
1893
基于OpenLake的Flink+Paimon+EMR StarRocks流式湖仓分析
291
免费公测｜阿里云EMR Serverless StarRocks 公测正式开启！
6836
面向未来的开源 OLAP 技术架构探讨以及选型实践
9642
小红书 API 接口使用指南：笔记详情数据接口的接入与使用
568
有奖实践，基于EMR StarRocks实现游戏玩家画像和行为分析
27
展开全部
