scalactic

from multiprocessing import Pool, Process, Manager
import time


def writer(queue):
    """ Queue writer worker """
    # Produce data into the queue
    for i in range(10000):
        t = time.time()
        print({'message': i, 'time': t, 'queue_size': queue.qsize()})

scalactic

from multiprocessing import Process, Value


def func(val):
    val.value = val.value * 2
    return True


if __name__ == '__main__':
    val = Value("l", 1, lock=False)  # "l" -> typecode_or_type , 1 -> the value.

scalactic

from multiprocessing import Pool, Process, Manager
import time
import logging
import os


logging.basicConfig(level=logging.INFO, format='[%(levelname)s] [%(asctime)s] [%(process)s] %(message)s', datefmt='%d/%m/%Y %I:%M:%S %p')


def writer(queue):

scalactic

// Base64 encode
new String(java.util.Base64.getEncoder.encode("hello".getBytes()),java.nio.charset.StandardCharsets.UTF_8)

// Base64 decode
new String(java.util.Base64.getDecoder.decode("aGVsbG8=".getBytes()),java.nio.charset.StandardCharsets.UTF_8)

scalactic

import javax.crypto.{Cipher, KeyGenerator}
import javax.crypto.spec.SecretKeySpec
import java.util.Base64

val keyGenerator = KeyGenerator.getInstance("AES")
keyGenerator.init(128)

val secretKey = keyGenerator.generateKey
val encodedKey = Base64.getEncoder.encodeToString(secretKey.getEncoded)
val decodedKey = Base64.getDecoder.decode(encodedKey)

scalactic

//spark-shell --conf spark.sql.sources.partitionOverwriteMode=dynamic
import org.apache.spark.sql.SaveMode

val data = Seq((1,2,"20210701"),(1,3,"20210701"),(3,4,"20210702"),(3,5,"20210702"))
val df = spark.createDataFrame(data).toDF("col_1", "col_2", "prt_date")

// this will create the table, with partitions 20210701, 20210702
df
  .write
  .partitionBy("prt_date")

scalactic

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

val schema = StructType(
  Array(
    StructField("key", StringType, nullable = false),
    StructField("time", IntegerType, nullable = false)
  )
)

scalactic

function merge(arr, l, m, r) {
  let i, j, k;
  let n1 = m - l + 1;
  let n2 = r - m;
  /* create temp arrays */
  let L = new Array(n1);
  let R = new Array(n2);
  /* Copy data to temp arrays L[] and R[] */
  for (i = 0; i < n1; i++)
    L[i] = arr[l + i];

scalactic

import org.apache.spark.sql.types.{StructType, ArrayType}
import org.apache.spark.sql.catalyst.ScalaReflection

/** Simple schema */
case class A(key: String, time: java.sql.Timestamp, date: java.sql.Date, decimal: java.math.BigDecimal, map: Map[String, Int], nested: Seq[Map[String, Seq[Int]]])
val schema = ScalaReflection.schemaFor[A].dataType.asInstanceOf[StructType]
schema.printTreeString

/** Array schema */
val arrSchema = ScalaReflection.schemaFor[Seq[A]].dataType.asInstanceOf[ArrayType]

scalactic

import java.sql.Timestamp
import java.util.logging.{ConsoleHandler, Formatter, Level, LogRecord, Logger}

trait ILogging {
  private def setupLogger(level: Level): Logger = {
    object SimpleFormatter extends Formatter {
      override def format(record: LogRecord): String =
        s"[${new Timestamp(record.getMillis)}] [${record.getLevel}] [${record.getLoggerName}] ${record.getMessage} \n"
    }
    val logger: Logger = Logger.getLogger(this.getClass.getName)