brachi-wernick

public class FieldsConfigurationProvider  {

    private Logger logger = LoggerFactory.getLogger(FieldsConfigurationProvider.class);
    private Long lastUpdateTime;
    private FieldsConfiguration config;

    private static FieldsConfigurationProvider instance = null;
    public static String bucketName;
    public static  String filePath;

brachi-wernick

WriteResult writeResult = tableRowToInsertCollection
        .apply("BQ-write", BigQueryIO.write()
        // specify that failed rows will be returned with their error
                .withExtendedErrorInfo()
                .to(tableSpec)
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
        //Specfies a policy for handling failed inserts.
                .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors()));
                

brachi-wernick

EventsProcessingOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
                    .as(EventsProcessingOptions.class);

Pipeline p = Pipeline.create(options);

PCollectionTuple tableRows =
                  // read kafka topic
                   p.apply("kafka-topic-read", kafkaReader)
                    .apply("kafka-values", MapElements.into(TypeDescriptors.strings())
                            .via(record ->record.getKV().getValue()))

brachi-wernick

@ProcessElement
public void processElement(@Element JsonNode> element, OutputReceiver<TableRowWithEvent> out) {
  
  public static final TupleTag<JsonNode> MAIN_OUT = new TupleTag<JsonNode>() {};
  public static final TupleTag<BigQueryProcessError> DEADLETTER_OUT = new TupleTag<BigQueryProcessError>() {};
  
  TableRow convertedRow = new TableRow();

  try {
    

brachi-wernick

@ProcessElement
public void processElement(@Element JsonNode> element, OutputReceiver<TableRowWithEvent> out) {
  
  TableRow convertedRow = new TableRow();

  insertLong(element.get("server_time"), "server_time", convertedRow);
  insertFloat(element.get("screen_dpi"), "screen_dpi", convertedRow);
  
  // more transformation to come

brachi-wernick

p.apply("BQ-write", BigQueryIO.write()
        .to(tableSpec)
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND);

brachi-wernick

 PCollection<Long> ticks = p
                    // Produce 1 "tick" per second
                    .apply(GenerateSequence.from(0).withRate(1, Duration.standardSeconds(1)))
                    // Window the ticks into 1-minute windows
                    .apply(Window.into(FixedWindows.of(Duration.standardSeconds(1))))
                    // Use an arbitrary per-window combiner to reduce to 1 element per window
                    .apply(Combine.globally(Count.<Long>combineFn()).withoutDefaults());

            String bucketName = options.getBigQuerySchemaConfigBucketName().get();

brachi-wernick

brachi-wernick

private static GenericRecord buildRecord(LoginAttackCount loginAttackCount) throws IOException {
    // avro schema avsc file path.
    String schemaPath = "src/main/java/loginStream/login-attack-count.avsc";

    String schemaString;

    try (FileInputStream inputStream = new FileInputStream(schemaPath)) {
        schemaString = org.apache.commons.io.IOUtils.toString(inputStream);
    } catch (Exception e) {
        throw new RuntimeException(e);

brachi-wernick

counts.mapValues((windowed, counter) ->
{
    try {
        return buildRecord(new LoginAttackCount(
                windowed.key(),
                counter,
                windowed.window().start(),
                windowed.window().end()));
    } catch (IOException e) {
        throw new RuntimeException(e);