11gR2 New Feature – File Watchers

The File Watcher is a scheduler object that starts a job whenever files whose attributes met the defined criteria arrived on a system. These criteria include the name, location, and other properties of a file. When the file watcher detects the arrival of the designated file, it raises a file-arrival event. The event message, which has all information on the newly-arrived file, can then be used to process the file.

This new feature simplifies the configurations of the most common triggering event in the data load/batch processing which is to detect the arrival of files.

File Watcher configuration

Setup a new database account to manage the file watcher.

SQL> create user watcher_user identified by watcher_pwd
quota unlimited on users;
User created

SQL> grant connect to watcher_user;
Grant succeeded.

SQL> grant EXECUTE on SYS.SCHEDULER_FILEWATCHER_RESULT to watcher_user;
Grant succeeded.

Other grants needed to complete the tests:

grant create table, create procedure, create job to watcher_user;
grant execute on dbms_lock to watcher_user;
grant execute on dbms_system to watcher_user;
grant manage scheduler to watcher_user;

SQL> create or replace directory STAGING_DIR as '/home/oracle/staging';
Directory created.

SQL> grant read, write on directory staging_dir to watcher_user;
Grant succeeded.

Now as a new watcher_user, we will configure the File Watcher.

1. Create a credential using the OS privilege for file access.

begin
  dbms_scheduler.create_credential(
  credential_name => 'watch_credential',
  username => 'oracle',
  password => 'oracle');
end;
/

2. Create a table to store data processed from file.

create table t_staging_files(
  upload_timestamp  timestamp,
  file_name         varchar2(100),
  file_size         number,
  contents          clob
);

3. The procedure will process file data and put into a database table.

create or replace procedure process_files
(payload IN sys.scheduler_filewatcher_result)
is
  l_clob clob;
  l_bfile bfile;

  dest_offset  INTEGER := 1;
  src_offset   INTEGER := 1;
  src_csid     NUMBER  := NLS_CHARSET_ID ('AL32UTF8');
  lang_context INTEGER := dbms_lob.default_lang_ctx;
  warning      INTEGER;
begin
  insert into t_staging_files (
    upload_timestamp , file_name, file_size, contents)
  values(
    payload.file_timestamp,
    payload.directory_path || '/' || payload.actual_file_name,
    payload.file_size,
    empty_clob()
  ) returning contents into l_clob;

  l_bfile := bfilename('STAGING_DIR', payload.actual_file_name);
  dbms_lob.fileopen(l_bfile);
  dbms_lob.loadclobfromfile (
    l_clob,
    l_bfile,
    dbms_lob.getlength(l_bfile),
    dest_offset,
    src_offset,
    src_csid,
    lang_context,
    warning
  );
  dbms_lob.fileclose(l_bfile);
end;
/

4. Create a Program object with a Metadata argument.

begin
  dbms_scheduler.create_program (
    program_name        => 'file_watcher',
    program_type        => 'stored_procedure',
    program_action      => 'process_files',
    number_of_arguments => 1,
    enabled             => false);

  dbms_scheduler.define_metadata_argument (
    program_name        => 'file_watcher',
    metadata_attribute  => 'event_message',
    argument_position   => 1);

  dbms_scheduler.enable('file_watcher');

end;
/
PL/SQL procedure successfully completed.

5. Create a File Watcher

begin
  dbms_scheduler.create_file_watcher(
    file_watcher_name => 'my_file_watcher',
    directory_path    => '/home/oracle/staging',
    file_name         => '*',
    credential_name   => 'watch_credential',
    destination       => null,
    enabled           => false);
end;
/
PL/SQL procedure successfully completed.

6. Create an Event-Based Job that references the File Watcher.

begin
  dbms_scheduler.create_job(
    job_name        => 'staging_file_job',
    program_name    => 'file_watcher',
    event_condition => 'tab.user_data.file_size > 10',
    queue_spec      => 'my_file_watcher',
    auto_drop       => false,
    enabled         => false);

    dbms_scheduler.set_attribute('staging_file_job','parallel_instances',true);
end;
/

7. Enable all objects

begin
  dbms_scheduler.enable('my_file_watcher,staging_file_job');
end;
/

8. Perform validation

$ echo "Hello World Hello World" > /home/oracle/staging/test_file.txt

After waiting for about 10-15 minutes,

col UPLOAD_TIMESTAMP format a20
col FILE_NAME format a20
col CONTENTS format a20

select * from t_staging_files;

UPLOAD_TIMESTAMP     FILE_NAME             FILE_SIZE CONTENTS
-------------------- -------------------- ---------- -----------------------

13-OCT-09 01.42.04.0 /home/oracle/staging         23 Hello World Hello World
00000 PM             /test_file.txt

By default, the file watcher checks for the arrival of files every 10 minutes. You can adjust this interval as follows:

as SYS user

begin
 DBMS_SCHEDULER.SET_ATTRIBUTE('FILE_WATCHER_SCHEDULE','REPEAT_INTERVAL','FREQ=MINUTELY;INTERVAL=2');
end;
/

You can view information about file watchers by querying the views *_SCHEDULER_FILE_WATCHERS.

col FILE_WATCHER_NAME format a20
col DIRECTORY_PATH format a20
col FILE_NAME format a5
col CREDENTIAL_NAME format a17

SELECT file_watcher_name, directory_path, file_name, credential_name
FROM dba_scheduler_file_watchers;

FILE_WATCHER_NAME    DIRECTORY_PATH       FILE_ CREDENTIAL_NAME
-------------------- -------------------- ----- -----------------
MY_FILE_WATCHER      /home/oracle/staging *     WATCH_CREDENTIAL

References:

Oracle 11gR2 document: Starting a Job When a File Arrives on a System