timvw

#!/bin/bash
# Export variables if not set already
export CLOUDKARAFKA_BROKERS="${CLOUDKARAFKA_BROKERS:-ark-01.srvs.cloudkafka.com:9094,ark-02.srvs.cloudkafka.com:9094,ark-03.srvs.cloudkafka.com:9094}"
export CLOUDKARAFKA_USERNAME="${CLOUDKARAFKA_USERNAME:-herecomesyourusername}"
export CLOUDKARAFKA_PASSWORD="${CLOUDKARAFKA_PASSWORD:-herecomesyourpassword}"

timvw

import java.io._
import scala.xml.XML
import ammonite.ops._
import ammonite.ops.ImplicitWd._

val tempFile = File.createTempFile("tmp", "tmp")
tempFile.deleteOnExit()
val f = tempFile.getAbsolutePath
val res = %%('mvn, "help:effective-pom", s"-Doutput=${f}")
val xml = XML.loadFile(f)

timvw

trait IHaveLakeName {
  def lakeName: String
}

trait IHaveStreamName {
  def streamName: String
}

trait IHaveLakeAndStreamName extends IHaveLakeName with IHaveStreamName

timvw

    import org.apache.spark.sql.SparkSession
    import org.apache.spark.sql.cassandra._
    import com.datastax.spark.connector.cql.CassandraConnectorConf.ConnectionHostParam
    import com.datastax.spark.connector.cql.DefaultAuthConfFactory.{PasswordParam, UserNameParam}

    val spark = {
      SparkSession
        .builder()
        .master("local[*]")
        .appName("test")

timvw

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val ds = spark.read
      .format("kafka")
      .option("kafka.bootstrap.servers", bootstrapServers)
      .option("subscribe", topic)
      .option("startingOffsets", "earliest")
      .load()
      .orderBy(desc("timestamp"))

timvw

mvn deploy:deploy-file -DgroupId=com.sksamuel.avro4s \
    -DartifactId=avro4s-core_2.11 \
    -Dversion=2.0.1 \
    -Dpackaging=jar \
    -Dfile=/Users/timvw/Desktop/avro4s-core_2.11-2.0.1.jar \
    -DgeneratePom=true \
    -DrepositoryId=icteam \
    -Durl=http://nexus.icteam.be/nexus/content/repositories/custom/

timvw

    val groupId = "demo-consumer"
    val bootstrapServers = "localhost:9092"

    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("demo")
      //.config("spark.sql.streaming.stateStore.maintenanceInterval", "10s")
      //.config("spark.sql.streaming.metricsEnabled", true)
      .config("spark.sql.streaming.minBatchesToRetain", 2)

timvw

# download jars from nexus
mvn dependency:copy-dependencies

# copy jars into a folder under plugin.path (connect-distribute.properties)
# in this example plugin.path=./data/kafka/jars
cp -R ./target/dependency ./data/kafka/jars/kafka-connect-hdfs

timvw

    val spark = SparkSession
      .builder()
      .config("spark.sql.avro.compression.codec", "snappy")
      .appName("test")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    val file = "/Users/timvw/src/docker/docker-streaming/data/kafka/hdfs/topics/test/year=2018/month=06/day=06/hour=20/test+0+0000000000+0000000029.avro"

timvw

  val entityMetadataSchema = SchemaBuilder
    .record("entityMetadata")
    .fields()
    .name("relatedLookupRecords").`type`().array().items(Schema.create(Schema.Type.STRING)).noDefault()
    .endRecord()

  def serialize(x: GenericRecord, s: Schema): Array[Byte] = {
    val baos = new ByteArrayOutputStream()
    val fw = new DataFileWriter[GenericRecord](new GenericDatumWriter[GenericRecord]())
    fw.create(s, baos)