3

我需要为 Hive 数据库中的每个表计算一些聚合。我的代码是这样的： 

sc = SparkContext()
sqlContext = HiveContext(sc)

showtables_df = sqlContext.sql('show tables in my_db')
for onlinetable in showtables_df.select('tableName').rdd.collect():
    hive_table_name = onlinetable['tableName']
    try:
        table_df = sqlContext.sql('SELECT * FROM my_db.' + hive_table_name)
        table_df.count() # sample action
    except Exception as e:
        logger.error('Error in table ' + hive_table_name)
        logger.error(type(e))
        logger.error(e)

在某个时刻，巨大的表读取会引发异常，并且 SparkContext 会关闭。从此时起，对 sqlContext 的每次调用都会失败： 

[2018-05-20 07:40:36] ERROR - my_table: An error occurred while calling o40.sql.
: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
This stopped SparkContext was created at:

org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:59)
sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
java.lang.reflect.Constructor.newInstance(Constructor.java:423)
py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:234)
py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
py4j.Gateway.invoke(Gateway.java:214)
py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:79)
py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:68)
py4j.GatewayConnection.run(GatewayConnection.java:209)
java.lang.Thread.run(Thread.java:745)

The currently active SparkContext was created at:

org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:59)
sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
java.lang.reflect.Constructor.newInstance(Constructor.java:423)
py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:234)
py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
py4j.Gateway.invoke(Gateway.java:214)
py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:79)
py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:68)
py4j.GatewayConnection.run(GatewayConnection.java:209)
java.lang.Thread.run(Thread.java:745)

        at org.apache.spark.SparkContext.org$apache$spark$SparkContext$$assertNotStopped(SparkContext.scala:106)
        at org.apache.spark.SparkContext$$anonfun$parallelize$1.apply(SparkContext.scala:729)
        at org.apache.spark.SparkContext$$anonfun$parallelize$1.apply(SparkContext.scala:728)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
        at org.apache.spark.SparkContext.withScope(SparkContext.scala:714)
        at org.apache.spark.SparkContext.parallelize(SparkContext.scala:728)
        at org.apache.spark.sql.execution.ExecutedCommand.doExecute(commands.scala:70)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
        at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
        at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:55)
        at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:55)
        at org.apache.spark.sql.DataFrame.<init>(DataFrame.scala:145)
        at org.apache.spark.sql.DataFrame.<init>(DataFrame.scala:130)
        at org.apache.spark.sql.DataFrame$.apply(DataFrame.scala:52)
        at org.apache.spark.sql.SQLContext.sql(SQLContext.scala:817)
        at sun.reflect.GeneratedMethodAccessor23.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:231)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
        at py4j.Gateway.invoke(Gateway.java:259)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:209)
        at java.lang.Thread.run(Thread.java:745)

截至目前，我无法为我的工作添加资源，但我想重新启动 SparkContext 并继续在 for 循环的剩余表中计算聚合。

检查 SparkContext 是否在异常处理程序中关闭并最终重新创建它是一个好（且可行）的想法： 

...
except Exception as e:
    logger.error('Error in table ' + hive_table_name)
    logger.error(type(e))
    logger.error(e)
    if sc._jsc.sc().isStopped():
        sc = SparkContext()
        sqlContext = HiveContext(sc)

我正在使用 Spark 1.6.1 版，使用--master yarn --deploy-mode 客户端运行作业

4

0 回答 0