update the develop branch with items from main

.github/workflows/almaren-framework.yml

@@ -0,0 +1,42 @@
+name: Almaren Framework
+on: [push, pull_request]
+
+jobs:
+  Build:
+    runs-on: ubuntu-20.04
+    services:
+      postgres:
+        image: postgres:13.4
+        env:
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_HOST_AUTH_METHOD: trust
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+      - name : Check out repository code
+        uses: actions/checkout@v2
+      - name: Setup JDK
+        uses: actions/setup-java@v3
+        with:
+            distribution: temurin
+            java-version: 8
+            cache: sbt
+
+      - name: Build and test scala version
+        run: |
+          PGPASSWORD="postgres" psql -c 'create database almaren;' -U postgres -h localhost 
+          PGPASSWORD="postgres" psql -c "ALTER USER postgres PASSWORD 'postgres' ;" -U postgres -h localhost 
+          PGPASSWORD="postgres" psql -c 'create role runner;' -U postgres -h localhost
+          PGPASSWORD="postgres" psql -c 'ALTER ROLE "runner" WITH  LOGIN SUPERUSER INHERIT CREATEDB CREATEROLE REPLICATION;' -U postgres -h localhost
+          sbt ++2.12.15 test
+          sbt ++2.13.9 test
+          rm -rf "$HOME/.ivy2/local" || true
+          find $HOME/Library/Caches/Coursier/v1        -name "ivydata-*.properties" -delete || true
+          find $HOME/.ivy2/cache                       -name "ivydata-*.properties" -delete || true
+          find $HOME/.cache/coursier/v1                -name "ivydata-*.properties" -delete || true
+          find $HOME/.sbt                              -name "*.lock"               -delete || true

.gitignore

@@ -1,23 +1,20 @@
-# vscode
-.vscode/
 
 # metals
 .metals/
 .bloop/
 project/metals.sbt
+project/project/
 .bsp/
 .bloop/
 target/
-project/project/
 
 # spark
 spark-warehouse
 
 metastore_db
 .gradle
 .vertx
-.metals
-.bloop
+.vscode
 
 # mac
 .DS_Store
@@ -140,3 +137,4 @@ flycheck_*.el
 # directory configuration
 .dir-locals.el
 
+

README.md

@@ -2,13 +2,15 @@
 
 The Almaren Framework provides a simplified consistent minimalistic layer over Apache Spark, while still allowing you to take advantage of native Apache Spark features. You can even combine it with standard Spark code.
 
-[![Build Status](https://travis-ci.com/mantovani/almaren-framework.svg?branch=master)](https://travis-ci.com/mantovani/almaren-framework)
+[![Build Status](https://github.com/music-of-the-ainur/almaren-framework/actions/workflows/almaren-framework.yml/badge.svg)](https://github.com/music-of-the-ainur/almaren-framework/actions/workflows/almaren-framework.yml)
 [![Gitter Community](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/music-of-the-ainur/community)
 
+
 ## Table of Contents
 
 - [Introduction](#introduction)
   * [Dependency](#dependency)
+  * [Connector Usage](#connector-usage)
   * [Batch Example](#batch-example)
   * [Streaming Example](#streaming-example)
 - [Debugging](#debugging)
@@ -41,6 +43,7 @@ The Almaren Framework provides a simplified consistent minimalistic layer over A
     + [targetSolr](#targetsolr)
     + [targetMongoDb](#targetmongodb)
     + [targetBigQuery](#targetbigquery)
+    + [targetFile](#targetfile)
 - [Executors](#executors)
   * [Batch](#batch)
   * [Streaming Kafka](#streaming-kafka)
@@ -59,14 +62,35 @@ The Almaren Framework provides a simplified consistent minimalistic layer over A
 To add Almaren Framework dependency to your sbt build:
 
 ```
-libraryDependencies += "com.github.music-of-the-ainur" %% "almaren-framework" % "0.9.3-3.1"
+libraryDependencies += "com.github.music-of-the-ainur" %% "almaren-framework" % "0.9.11-3.5"
 ```
 
 To run in spark-shell:
 
+For scala version(2.12):
+```
+spark-shell --packages "com.github.music-of-the-ainur:almaren-framework_2.12:0.9.11-3.5"
 ```
-spark-shell --packages "com.github.music-of-the-ainur:almaren-framework_2.12:0.9.3-3.1"
+For scala version(2.13):
 ```
+spark-shell --packages "com.github.music-of-the-ainur:almaren-framework_2.13:0.9.11-3.5"
+```
+Almaren connector is available in
+[Maven Central](https://mvnrepository.com/artifact/com.github.music-of-the-ainur) repository.
+
+| version                    | Connector Artifact                                                |
+|----------------------------|-------------------------------------------------------------------|
+| Spark 3.5.x and scala 2.13 | `com.github.music-of-the-ainur:almaren-framework_2.13:0.9.11-3.5` |
+| Spark 3.5.x and scala 2.12 | `com.github.music-of-the-ainur:almaren-framework_2.12:0.9.11-3.5` |
+| Spark 3.4.x and scala 2.13 | `com.github.music-of-the-ainur:almaren-framework_2.13:0.9.11-3.4` |
+| Spark 3.4.x and scala 2.12 | `com.github.music-of-the-ainur:almaren-framework_2.12:0.9.11-3.4` |
+| Spark 3.3.x and scala 2.13 | `com.github.music-of-the-ainur:almaren-framework_2.13:0.9.11-3.3` |
+| Spark 3.3.x and scala 2.12 | `com.github.music-of-the-ainur:almaren-framework_2.12:0.9.11-3.3` |
+| Spark 3.2.x and scala 2.12 | `com.github.music-of-the-ainur:almaren-framework_2.12:0.9.11-3.2` |
+| Spark 3.1.x and scala 2.12 | `com.github.music-of-the-ainur:almaren-framework_2.12:0.9.11-3.1` |
+| Spark 2.4.x and scala 2.12 | `com.github.music-of-the-ainur:almaren-framework_2.12:0.9.11-2.4` |
+| Spark 2.4.x and scala 2.11 | `com.github.music-of-the-ainur:almaren-framework_2.11:0.9.11-2.4` |
+
 
 ### Batch Example
 ```scala
@@ -80,6 +104,7 @@ val almaren = Almaren("App Name")
 val spark = almaren.spark
     .master("local[*]")
     .config("spark.sql.shuffle.partitions", "1")
+    .getOrCreate
 
 val df:DataFrame = almaren.builder
     .sourceSql("select monotonically_increasing_id() as id,* from movies")
@@ -90,8 +115,8 @@ val df:DataFrame = almaren.builder
         |cast[1]$support_actor:StringType
         |genres[0]$genre:StringType
         |director@director
-        |	director.name$credit_name:StringType""".stripMargin)
-    .sql("""SELECT * FROM __TABLE__ WHERE actor NOT IN ("the","the life of")""")
+        |	director.name$credit_name:StringType""".stripMargin).alias("table")
+    .sql("""SELECT * FROM table WHERE actor NOT IN ("the","the life of")""")
     .targetJdbc("jdbc:postgresql://localhost/almaren","org.postgresql.Driver","movies",SaveMode.Overwrite)
     .batch
 ```
@@ -119,9 +144,9 @@ val streaming = almaren.builder
         |text$message:StringType
         |entities@entitie
         |	entitie.hashtags@hashtag
-        |		hashtag.text$hashtag:StringType""".stripMargin)
-  .sql("SELECT DISTINCT * FROM __TABLE__")
-  .sql("""SELECT sha2(concat_ws("",array(*)),256) as unique_hash,*,current_timestamp from __TABLE__""")
+        |		hashtag.text$hashtag:StringType""".stripMargin).alias("table")
+  .sql("SELECT DISTINCT * FROM table").alias("table1")
+  .sql("""SELECT sha2(concat_ws("",array(*)),256) as unique_hash,*,current_timestamp from table1""")
   .targetJdbc("jdbc:postgresql://localhost/almaren","org.postgresql.Driver","twitter_streaming",SaveMode.Append)
 
 almaren.streaming(streaming,Map("kafka.bootstrap.servers" -> "localhost:9092","subscribe" -> "twitter", "startingOffsets" -> "earliest"))
@@ -150,8 +175,8 @@ val df:DataFrame = almaren.builder
         |cast[1]$support_actor:StringType
         |genres[0]$genre:StringType
         |director@director
-        |	director.name$credit_name:StringType""".stripMargin)
-    .sql("""SELECT *,current_timestamp as date FROM __TABLE__ WHERE actor NOT IN ("the","the life of")""")
+        |	director.name$credit_name:StringType""".stripMargin).alias("table")
+    .sql("""SELECT *,current_timestamp as date FROM table WHERE actor NOT IN ("the","the life of")""")
     .targetJdbc("jdbc:postgresql://localhost/almaren","org.postgresql.Driver","movies",SaveMode.Overwrite)
     .batch
 ```
@@ -336,6 +361,12 @@ Cache/Uncache both DataFrame or Table
 cache(true)
 ```
 
+Cache Dataframe with Storage Level
+
+```scala
+cache(true,storageLevel = MEMORY_AND_DISK)
+```
+
 #### Coalesce
 
 Decrease the number of partitions in the RDD to numPartitions. Useful for running operations more efficiently after filtering down a large dataset.
@@ -481,6 +512,14 @@ Write to BigQuery using [BigQuery Connector](https://github.com/music-of-the-ain
 
 Write to Neo4j using [Neo4j Connector](https://github.com/music-of-the-ainur/neo4j.almaren)
 
+#### targetFile
+
+Write to File, you must have the following parameters: format, path, saveMode of the file and parameters as a Map. For partitioning provide a list of columns, for bucketing provide number of buckets and list of columns, for sorting provide list of columns, and tableName. Check the [documentation](https://spark.apache.org/docs/latest/sql-data-sources-generic-options.html) for the full list of parameters.
+
+```scala
+targetFile("parquet","/home/abc/targetlocation/output.parquet",SaveMode.Overwrite,Map("batchSize"->10000),List("partitionColumns"),(5,List("bucketingColumns")),List("sortingColumns"),Some("sampleTableName"))
+```
+
 ## Executors
 
 Executors are responsible to execute Almaren Tree i.e ```Option[Tree]``` to Apache Spark. Without invoke an _executor_, code won't be executed by Apache Spark. Follow the list of _executors_:
@@ -499,8 +538,8 @@ val tree = almaren.builder
         |cast[1]$support_actor:StringType
         |genres[0]$genre:StringType
         |director@director
-        |	director.name$credit_name:StringType""".stripMargin)
-    .sql("""SELECT * FROM __TABLE__ WHERE actor NOT IN ("the","the life of")""")
+        |	director.name$credit_name:StringType""".stripMargin).alias("table")
+    .sql("""SELECT * FROM table WHERE actor NOT IN ("the","the life of")""")
     .targetJdbc("jdbc:postgresql://localhost/almaren","org.postgresql.Driver","movies",SaveMode.Overwrite)
 
 val df:DataFrame = tree.batch
@@ -537,9 +576,9 @@ val tree = almaren.builder
         |text$message:StringType
         |entities@entitie
         |	entitie.hashtags@hashtag
-        |		hashtag.text$hashtag:StringType""".stripMargin)
-  .sql("SELECT DISTINCT * FROM __TABLE__")
-  .sql("""SELECT sha2(concat_ws("",array(*)),256) as unique_hash,*,current_timestamp from __TABLE__""")
+        |		hashtag.text$hashtag:StringType""".stripMargin).alias("table")
+  .sql("SELECT DISTINCT * FROM table").alias("table1")
+  .sql("""SELECT sha2(concat_ws("",array(*)),256) as unique_hash,*,current_timestamp from table1""")
   .targetJdbc("jdbc:postgresql://localhost/almaren","org.postgresql.Driver","twitter_streaming",SaveMode.Append)
 
 almaren.streaming(tree,Map("kafka.bootstrap.servers" -> "localhost:9092","subscribe" -> "twitter", "startingOffsets" -> "earliest"))
@@ -578,14 +617,14 @@ Default value of sample ratio is 1.0
 val almaren = Almaren("appName")
 val df:DataFrame = almaren.builder.sourceSql("SELECT * FROM db.schema.table")
     .deserializer("JSON","json_str")
-    .dsl("uuid$id:StringType
+    .dsl("""uuid$id:StringType
         |code$area_code:LongType
         |names@name
         |	name.firstName$first_name:StringType
         |	name.secondName$second_name:StringType
         |	name.lastName$last_name:StringType
-        |source_id$source_id:LongType".stripMargin)
-    .sql("""SELECT *,unix_timestamp() as timestamp from __TABLE__""")
+        |source_id$source_id:LongType""".stripMargin).alias("table")
+    .sql("""SELECT *,unix_timestamp() as timestamp from table""")
     .targetJdbc("jdbc:postgresql://localhost/database","org.postgresql.Driver","target_table",SaveMode.Append)
 ```
 
@@ -596,22 +635,22 @@ val df:DataFrame = almaren.builder.sourceSql("SELECT * FROM db.schema.table")
 ```scala
 val almaren = Almaren("appName")
 
-val target1 = almaren.builder.dsl("uuid$id:StringType
+val target1 = almaren.builder.dsl("""uuid$id:StringType
     |code$area_code:LongType
     |names@name
     |    name.firstName$first_name:StringType
     |    name.secondName$second_name:StringType
     |    name.lastName$last_name:StringType
-    |source_id$source_id:LongType".stripMargin)
-.sql("SELECT *,unix_timestamp() as timestamp from __TABLE__")
+    |source_id$source_id:LongType""".stripMargin).alias("table")
+.sql("SELECT *,unix_timestamp() as timestamp from table")
 .targetCassandra("test1","kv1")
 
-val target2 = almaren.builder.dsl("uuid$id:StringType
+val target2 = almaren.builder.dsl("""uuid$id:StringType
     |code$area_code:LongType
     |phones@phone
     |    phone.number$phone_number:StringType
-    |source_id$source_id:LongType".stripMargin)
-.sql("SELECT *,unix_timestamp() as timestamp from __TABLE__")
+    |source_id$source_id:LongType""".stripMargin).alias("table")
+.sql("SELECT *,unix_timestamp() as timestamp from table")
 .targetCassandra("test2","kv2")
 
 almaren.builder.sourceSql("SELECT * FROM db.schema.table")
@@ -636,7 +675,8 @@ val sourcePolicy = almaren.builder.sourceHbase("""{
     |"status":{"cf":"Policy", "col":"status", "type":"string"},
     |"person_id":{"cf":"Policy", "col":"source", "type":"long"}
     |}
-|}""").sql(""" SELECT * FROM __TABLE__ WHERE status = "ACTIVE" """).alias("policy")
+|}""").alias("hbase")
+        .sql(""" SELECT * FROM hbase WHERE status = "ACTIVE" """).alias("policy")
 
 val sourcePerson = almaren.builder.sourceHbase("""{
     |"table":{"namespace":"default", "name":"person"},
@@ -647,12 +687,13 @@ val sourcePerson = almaren.builder.sourceHbase("""{
     |"type":{"cf":"Policy", "col":"type", "type":"string"},
     |"age":{"cf":"Policy", "col":"source", "type":"string"}
     |}
-|}""").sql(""" SELECT * FROM __TABLE__ WHERE type = "PREMIUM" """).alias("person")
+    |}""").alias("hbase")
+        .sql(""" SELECT * FROM hbase WHERE type = "PREMIUM" """).alias("person")
 
-almaren.builder.sql(""" SELECT * FROM person JOIN policy ON policy.person_id = person.id """)
-    .sql("SELECT *,unix_timestamp() as timestamp FROM __TABLE__")
+almaren.builder.sql(""" SELECT * FROM person JOIN policy ON policy.person_id = person.id """).alias("table")
+    .sql("SELECT *,unix_timestamp() as timestamp FROM table").alias("table1")
     .coalesce(100)
-    .targetSql("INSERT INTO TABLE area.premimum_users SELECT * FROM __TABLE__")
+    .targetSql("INSERT INTO TABLE area.premimum_users SELECT * FROM table1")
     .batch(sourcePolicy,sourceHbase)
 ```
 
@@ -662,8 +703,8 @@ almaren.builder.sql(""" SELECT * FROM person JOIN policy ON policy.person_id = p
 
 ```scala
 val almaren = Almaren("appName")
-val sourceData = almaren.builder.sourceJdbc("oracle.jdbc.driver.OracleDriver","jdbc:oracle:thin:@localhost:1521:xe","SELECT * FROM schema.table WHERE st_date >= (sysdate-1) AND st_date < sysdate")
-    .sql("SELECT to_json(named_struct('id', id,))) as __BODY__ from __TABLE__")
+val sourceData = almaren.builder.sourceJdbc("oracle.jdbc.driver.OracleDriver","jdbc:oracle:thin:@localhost:1521:xe","SELECT * FROM schema.table WHERE st_date >= (sysdate-1) AND st_date < sysdate").alias("table")
+    .sql("SELECT to_json(named_struct('id', id,))) as __BODY__ from table")
     .coalesce(30)
     .targetHttp("https://host.com:9093/api/foo","post",Map("Authorization" -> "Basic QWxhZGRpbjpPcGVuU2VzYW1l"))