Re: [I] MetricsReporter support close [iceberg]

["allen-abhishekkr (via GitHub)" <gi...@apache.org>]

allen-abhishekkr commented on issue #9349:
URL: https://github.com/apache/iceberg/issues/9349#issuecomment-2153099416

   @huyuanfeng2018 
   
   I see that the "closeable" trait was added in this commit, however in case of below implementations, the close() method of the MetricsReporter never gets called. 
   
    spark-submit command looks like below:
   `
   spark-submit --name test_metrics --packages org.apache.iceberg:iceberg-aws-bundle:1.5.0,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0  \
   --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
   --conf spark.sql.catalog.prod=org.apache.iceberg.spark.SparkCatalog \
   --conf spark.sql.catalog.prod.type=hive \
   --conf spark.sql.catalog.prod.uri=thrift://<host>:9083 \
   --conf spark.sql.catalog.prod.warehouse=s3a://<mybucket>/jarSparkJobs/iceberg-metrics-test/ \
   --conf spark.sql.catalog.prod.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
   --conf iceberg.engine.hive.enabled=true \
   --conf spark.dynamicAllocation.enabled=false \
   --conf spark.sql.catalog.prod.metrics-reporter-impl=package.MyCustomMetricsReporter \
   --class in.app.TestMetrics /home/hadoop/abhishek/my_jar.jar
   `
   
   The MetricsReporter implementation as below:
   `
   class MyCustomMetricsReporter extends MetricsReporter{
    override def report(report: MetricsReport): Unit = {
     //gets called
     }
   
    override def close(): Unit = {
     //Never Gets Called
     }
   }
   `
   
   The Main Class
   `
   import org.apache.spark.sql.SparkSession
   import org.apache.spark.sql.types._
   import org.apache.spark.sql.Row
   
   import scala.util.Random
   
   object TestMetrics extends App {
   
     val spark: SparkSession = SparkSession.builder()
       .enableHiveSupport()
       .getOrCreate()
   
     val schema = StructType( Array(
       StructField("vendor_id", LongType,true),
       StructField("trip_id", LongType,true),
       StructField("trip_distance", FloatType,true),
       StructField("fare_amount", DoubleType,true),
       StructField("store_and_fwd_flag", StringType,true)
     ))
   
     val randData = Random.nextDouble()
     println(randData)
   
     val data = Seq(
       Row(1: Long, 1000371: Long, 1.8f: Float, 15.32: Double, "N": String),
       Row(2: Long, 1000372: Long, 2.5f: Float, 22.15: Double, "N": String),
       Row(2: Long, 1000373: Long, 0.9f: Float, randData: Double, "N": String),
       Row(1: Long, 1000374: Long, 8.4f: Float, 42.13: Double, "Y": String)
     )
   
     val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)
     df.createTempView("latestRecords")
     val mergeQuery =
       """
         |MERGE INTO prod.iceberg_sanity_test.taxis t USING latestRecords s
         |        ON s.vendor_id=t.vendor_id and s.trip_id=t.trip_id
         |   WHEN MATCHED THEN UPDATE SET *
         |   WHEN NOT MATCHED THEN INSERT *
         |""".stripMargin
     println(System.currentTimeMillis())
     spark.sql(mergeQuery)
   
   }
   `
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org