Cluster-mode job compute-time/cost metrics
Hello Spark experts - I’m running Spark jobs in cluster mode using a dedicated cluster for each job. Is there a way to see how much compute time each job takes via Spark APIs, metrics, etc.? In case it makes a difference, I’m using AWS EMR - I’d ultimately like to be able to say this job costs $X since it took Y minutes on Z instance types (assuming all of the nodes are the same instance type), but I figure I could probably need to get the Z instance type through EMR APIs. Thanks! Jack
Unsubscribe
Unsubscribe
Unsubscribe
Unsubscribe
unsubscribe
unsubscribe
Spark 3.1.3 with Hive dynamic partitions fails while driver moves the staged files
Hi all, Running on Dataproc 2.0/1.3/1.4, we use INSERT INTO OVERWRITE command to insert new (time) partitions into existing Hive tables. But we see too many failures coming from org.apache.hadoop.hive.ql.metadata.Hive.replaceFiles. This is where the driver moves the successful files from staging to final directory. For some reason, the underlying FS implementation - GoogleCloudStorageImpl in this case - fails to move at least one file and the exception is propagated all the way through. We see many different failures - from hadoop.FileSystem.mkdirs, rename, etc., all coming from Hive.replaceFiles(). I guess FS failures are expected, but nowhere in org.apache.hadoop.hive.ql.metadata.Hive.loadDynamicPartitions there is try/catch/retry mechanism. Is it expected that the FS implementation do it? Because the GCS connector does not seem to do that :) We ended up patching and rebuilding hive-exec jar as an immediate mitigation (try/catch/retry), while our platform teams are reaching out to GCP support. I know this is more of a Hive issue rather than Spark, but still I wonder if anybody has encountered this issue, or similar? Thanks, Shay
unsubscribe
-- Sergei Boitsov JetBrains GmbH Christoph-Rapparini-Bogen 23 80639 München Handelsregister: Amtsgericht München, HRB 187151 Geschäftsführer: Yury Belyaev
unsubscribe
-- “Overfitting” is not about an excessive amount of physical exercise...
Re: [EXTERNAL] Re: [EXTERNAL] Re: Spark-submit without access to HDFS
Hey Mich,
Thanks for the detailed response. I get most of these options.
However, what we are trying to do is avoid having to upload the source
configs and pyspark.zip files to the cluster every time we execute the job
using spark-submit. Here is the code that does it:
https://github.com/apache/spark/blob/bacdb3b5fec9783f4604276480eb2a0f5702/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala#L813
Wondering if there is a way to skip uploading the configs. Uploading the
pyspark.zip file can be skipped by setting
PYSPARK_ARCHIVES_PATH=local://
On Mon, Dec 11, 2023 at 5:15 AM Mich Talebzadeh
wrote:
> Hi Eugene,
>
> With regard to your points
>
> What are the PYTHONPATH and SPARK_HOME env variables in your script?
>
> OK let us look at a typical of my Spark project structure
>
> - project_root
> |-- README.md
> |-- __init__.py
> |-- conf
> | |-- (configuration files for Spark)
> |-- deployment
> | |-- deployment.yaml
> |-- design
> | |-- (design-related files or documentation)
> |-- othermisc
> | |-- (other miscellaneous files)
> |-- sparkutils
> | |-- (utility modules or scripts specific to Spark)
> |-- src
> |-- (main source code for your Spark application)
>
> If you want Spark to recognize modules from the sparkutils directory or
> any other directories within your project, you can include those
> directories in the PYTHONPATH.
>
> For example, if you want to include the sparkutils directory:
>
> export PYTHONPATH=/path/to/project_root/sparkutils:$PYTHONPATH
> to recap, the ${PYTHONPATH} variable is primarily used to specify
> additional directories where Python should look for modules and packages.
> In the context of Spark, it is typically used to include directories
> containing custom Python code or modules that your Spark application
> depends on.
>
> With regard to
>
> The --conf "spark.yarn.appMasterEnv.SPARK_HOME=$SPARK_HOME" configuration
> option in Spark is used when submitting a Spark application to run on YARN
>
>-
>
>--conf: This is used to specify Spark configuration properties when
>submitting a Spark application.
>-
>
>spark.yarn.appMasterEnv.SPARK_HOME: This is a Spark configuration
>property that defines the value of the SPARK_HOME environment variable
>for the Spark application's Application Master (the process responsible for
>managing the execution of tasks on a YARN cluster).
>-
>
>$SPARK_HOME: This holds the path to the Spark installation directory.
>
> This configuration is setting the SPARK_HOME environment variable for the
> Spark Application Master when the application is running on YARN. This is
> important because the Spark Application Master needs to know the location
> of the Spark installation directory (SPARK_HOME) to configure and manage
> the Spark application's execution on the YARN cluster. HTH
> Mich Talebzadeh,
> Distinguished Technologist, Solutions Architect & Engineer
> London
> United Kingdom
>
>
>view my Linkedin profile
>
Re: [PySpark][Spark Dataframe][Observation] Why empty dataframe join doesn't let you get metrics from observation?
Hey Enrico it does help to understand it, thanks for explaining.
Regarding this comment
> PySpark and Scala should behave identically here
Is it ok that Scala and PySpark optimization works differently in this case?
вт, 5 дек. 2023 г. в 20:08, Enrico Minack :
> Hi Michail,
>
> with spark.conf.set("spark.sql.planChangeLog.level", "WARN") you can see
> how Spark optimizes the query plan.
>
> In PySpark, the plan is optimized into
>
> Project ...
> +- CollectMetrics 2, [count(1) AS count(1)#200L]
> +- LocalTableScan , [col1#125, col2#126L, col3#127, col4#132L]
>
> The entire join gets optimized away into an empty table. Looks like it
> figures out that df has no rows with col1 = 'c'. So df is never consumed
> / iterated, so the observation does not retrieve any metrics.
>
> In Scala, the optimization is different:
>
> *(2) Project ...
> +- CollectMetrics 2, [count(1) AS count(1)#63L]
> +- *(1) Project [col1#37, col2#38, col3#39, cast(null as int) AS
> col4#51]
> +- *(1) Filter (isnotnull(col1#37) AND (col1#37 = c))
>+- CollectMetrics 1, [count(1) AS count(1)#56L]
> +- LocalTableScan [col1#37, col2#38, col3#39]
>
> where the join also gets optimized away, but table df is still filtered
> for col1 = 'c', which iterates over the rows and collects the metrics for
> observation 1.
>
> Hope this helps to understand why there are no observed metrics for
> Observation("1") in your case.
>
> Enrico
>
>
>
> Am 04.12.23 um 10:45 schrieb Enrico Minack:
>
> Hi Michail,
>
> observations as well as ordinary accumulators only observe / process
> rows that are iterated / consumed by downstream stages. If the query
> plan decides to skip one side of the join, that one will be removed from
> the final plan completely. Then, the Observation will not retrieve any
> metrics and .get waits forever. Definitively not helpful.
>
> When creating the Observation class, we thought about a timeout for the
> get method but could not find a use case where the user would call get
> without first executing the query. Here is a scenario where though
> executing the query there is no observation result. We will rethink this.
>
> Interestingly, your example works in Scala:
>
> import org.apache.spark.sql.Observation
>
> val df = Seq(("a", 1, "1 2 3 4"), ("b", 2, "1 2 3 4")).toDF("col1",
> "col2", "col3")
> val df_join = Seq(("a", 6), ("b", 5)).toDF("col1", "col4")
>
> val o1 = Observation()
> val o2 = Observation()
>
> val df1 = df.observe(o1, count("*")).filter("col1 = 'c'")
> val df2 = df1.join(df_join, "col1", "left").observe(o2, count("*"))
>
> df2.show()
> +++++
> |col1|col2|col3|col4|
> +++++
> +++++
>
> o1.get
> Map[String,Any] = Map(count(1) -> 2)
>
> o2.get
> Map[String,Any] = Map(count(1) -> 0)
>
>
> Pyspark and Scala should behave identically here. I will investigate.
>
> Cheers,
> Enrico
>
>
>
> Am 02.12.23 um 17:11 schrieb Михаил Кулаков:
>
> Hey folks, I actively using observe method on my spark jobs and
> noticed interesting behavior:
> Here is an example of working and non working code:
> https://gist.github.com/Coola4kov/8aeeb05abd39794f8362a3cf1c66519c
>
>
>
> In a few words, if I'm joining dataframe after some filter rules and
> it became empty, observations configured on the first dataframe never
> return any results, unless some action called on the empty dataframe
> specifically before join.
>
> Looks like a bug to me, I will appreciate any advice on how to fix
> this behavior.
>
>
>
>
> -
> To unsubscribe e-mail: user-unsubscr...@spark.apache.org
>
>
>
Re: [EXTERNAL] Re: Spark-submit without access to HDFS
Hi Eugene,
With regard to your points
What are the PYTHONPATH and SPARK_HOME env variables in your script?
OK let us look at a typical of my Spark project structure
- project_root
|-- README.md
|-- __init__.py
|-- conf
| |-- (configuration files for Spark)
|-- deployment
| |-- deployment.yaml
|-- design
| |-- (design-related files or documentation)
|-- othermisc
| |-- (other miscellaneous files)
|-- sparkutils
| |-- (utility modules or scripts specific to Spark)
|-- src
|-- (main source code for your Spark application)
If you want Spark to recognize modules from the sparkutils directory or any
other directories within your project, you can include those directories in
the PYTHONPATH.
For example, if you want to include the sparkutils directory:
export PYTHONPATH=/path/to/project_root/sparkutils:$PYTHONPATH
to recap, the ${PYTHONPATH} variable is primarily used to specify
additional directories where Python should look for modules and packages.
In the context of Spark, it is typically used to include directories
containing custom Python code or modules that your Spark application
depends on.
With regard to
The --conf "spark.yarn.appMasterEnv.SPARK_HOME=$SPARK_HOME" configuration
option in Spark is used when submitting a Spark application to run on YARN
-
--conf: This is used to specify Spark configuration properties when
submitting a Spark application.
-
spark.yarn.appMasterEnv.SPARK_HOME: This is a Spark configuration
property that defines the value of the SPARK_HOME environment variable
for the Spark application's Application Master (the process responsible for
managing the execution of tasks on a YARN cluster).
-
$SPARK_HOME: This holds the path to the Spark installation directory.
This configuration is setting the SPARK_HOME environment variable for the
Spark Application Master when the application is running on YARN. This is
important because the Spark Application Master needs to know the location
of the Spark installation directory (SPARK_HOME) to configure and manage
the Spark application's execution on the YARN cluster. HTH
Mich Talebzadeh,
Distinguished Technologist, Solutions Architect & Engineer
London
United Kingdom
view my Linkedin profile
