Sample filter: embulk-filter-myapp

myapp.rb

module Embulk
  module Filter

    class MyappFilterPlugin < FilterPlugin
      Plugin.register_filter("myapp", self)

      def self.transaction(config, in_schema, &control)
        yield({}, in_schema)
      end

      def init
      end

      def close
      end

      def add(page)
        # find index of "comment" column
        columns = page.schema.select{|c| c.name == "comment" }
        idx = columns[0].index

        # filtering code:
        page.each do |record|
          case record[idx]
          when /^login failure:/
            record[idx] = "login failure"
          when /^login by/
            record[idx] = "login"
          end

          page_builder.add(record)
        end
      end

      def finish
        page_builder.finish
      end
    end

  end
end