Nick GrigorievNick
GrigorievNick
/ TestObservationWithForeach.scala
Last active
January 12, 2023 17:30
Spark Observation API Bug, reproduce
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import org.apache.spark.sql.Observation | |
| import org.apache.spark.sql.SparkSession | |
| import org.apache.spark.sql.functions._ | |
| import org.scalatest.FunSuite | |
| class TestObservationWithForeach extends FunSuite { | |
| test("observation") { | |
| implicit val spark: SparkSession = SparkSession.builder() |
GrigorievNick
/ SparkSQLGeneratePartitionOffset.scala
Last active
September 7, 2021 15:13
Spark Create unique sequential id per spark partition
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import org.apache.spark.sql.Column | |
| import org.apache.spark.sql.Row | |
| import org.apache.spark.sql.SparkSession | |
| import org.apache.spark.sql.catalyst.InternalRow | |
| import org.apache.spark.sql.catalyst.encoders.RowEncoder | |
| import org.apache.spark.sql.catalyst.expressions.LeafExpression | |
| import org.apache.spark.sql.catalyst.expressions.Stateful | |
| import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper | |
| import org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator | |
| import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext |
GrigorievNick
/ IcebergMergeOnRead.scala
Last active
August 16, 2021 21:29
IcebergRead from executor
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import com.google.common.collect.Maps | |
| import org.apache.iceberg.TableProperties | |
| import org.apache.iceberg.data.IcebergGenerics | |
| import org.apache.iceberg.expressions.Expressions | |
| import org.apache.iceberg.hadoop.HadoopTables | |
| import org.apache.iceberg.spark.SparkSchemaUtil.convert | |
| import org.apache.spark.TaskContext | |
| import org.apache.spark.sql.Row | |
| import org.apache.spark.sql.SaveMode | |
| import org.apache.spark.sql.SparkSession |
GrigorievNick
/ MergeDataframe.scala
Last active
August 4, 2021 09:49
Merge records in two dataframes by id columns
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import org.apache.spark.sql.catalyst.encoders.RowEncoder | |
| import org.apache.spark.sql.functions._ | |
| import org.apache.spark.sql.types.StructField | |
| import org.apache.spark.sql.types.StructType | |
| import org.apache.spark.sql.DataFrame | |
| import org.apache.spark.sql.Row | |
| import org.apache.spark.sql.SparkSession | |
| import org.scalatest.FunSuite | |
| import java.sql.Timestamp |
GrigorievNick
/ GracefulStopOnShutdownListener.scala
Last active
April 10, 2024 14:01
Spark Structure Streaming GraceFullShutdown on Sigterm. Sigterm will not interrupt currently running batch, but due to asynс nature of SparkQueryListner.onProgres method, can interrupt next batch during first few moments.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import org.apache.hadoop.util.ShutdownHookManager | |
| import org.apache.spark.sql.streaming.StreamingQueryListener | |
| import org.apache.spark.sql.streaming.StreamingQueryManager | |
| import org.slf4j.LoggerFactory | |
| import java.util.UUID | |
| import java.util.concurrent.ConcurrentHashMap | |
| import java.util.concurrent.SynchronousQueue | |
| import java.util.concurrent.TimeUnit |
GrigorievNick
/ ForceRepartitionByRangeCoalescePartitions.scala
Created
July 3, 2021 11:50
Spark custom Physical plan optimisation to AQE force post shuffle coalesce for repartitionByRange with user specified partitions num.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| package org.apache.spark.sql.execution.adaptive | |
| import org.apache.spark.sql.catalyst.rules.Rule | |
| import org.apache.spark.sql.execution.SparkPlan | |
| import org.apache.spark.sql.SparkSession | |
| import org.apache.spark.sql.catalyst.plans.physical.RangePartitioning | |
| import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec | |
| import org.apache.spark.sql.execution.CoalescedPartitionSpec | |
| import org.apache.spark.sql.internal.SQLConf |