audit_interval

`audit_interval`是一个操作符，用于在指定的时间间隔内对源Observable进行节流。它会忽略源Observable在这段时间内产生的所有值，只会发出最新的值。

以下是一个使用`audit_interval`的示例代码：

import { interval } from 'rxjs';
import { audit } from 'rxjs/operators';

interval(1000)
  .pipe(audit(() => interval(3000)))
 .subscribe(x => console.log(new Date(), x));

在这个示例中，我们创建了一个每秒发出一个值的Observable。然后，我们使用`audit`操作符，并传入一个返回每3秒发出一个值的Observable的函数作为参数。最后，我们订阅并打印出每次发出的值和当前时间。

这段代码的作用是，每当源Observable发出一个值时，`audit_interval`会等待3秒钟，然后发出这段时间内最新的值。这样可以确保在3秒内只发出最新的值，而忽略其他的值。

相关问题

create table #temp( audit_batch_id int null, audit_job_id int null, audit_src_sys_name nvarchar(50) null, audit_created_usr nvarchar(50) null, audit_updated_usr nvarchar(50) null, audit_created_tmstmp datetime null, audit_updated_tmstmp datetime null )，怎么给这个表初始化

这是一个 T-SQL 的语句，它创建了一个名为 `#temp` 的临时表，该表包含了七个列，分别是 `audit_batch_id`、`audit_job_id`、`audit_src_sys_name`、`audit_created_usr`、`audit_updated_usr`、`audit_created_tmstmp` 和 `audit_updated_tmstmp`，均为可空的数据类型。

要给这个表初始化，可以使用以下语句：

INSERT INTO #temp (audit_batch_id, audit_job_id, audit_src_sys_name, audit_created_usr, audit_updated_usr, audit_created_tmstmp, audit_updated_tmstmp) VALUES
(1, 1001, 'System A', 'User A', 'User B', '2021-01-01 00:00:00', '2021-01-01 01:00:00'),
(2, 1002, 'System B', 'User B', 'User C', '2021-01-02 00:00:00', '2021-01-02 01:00:00'),
(3, 1003, 'System C', 'User C', 'User A', '2021-01-03 00:00:00', '2021-01-03 01:00:00');

这个语句会向 `#temp` 表中插入三条记录，每条记录对应七个列的值，可以根据需要修改这些值来进行表的初始化。注意，临时表只在当前会话中存在，当会话结束时，临时表将被自动删除。

def add_audit_columns(self, dataframe, write_params): try: print(write_params) audit_source_sys_name = write_params.get("audit_src_sys_name", "") or write_params.get( "table_full_name") or write_params.get("data_source_name", "") print("audit_source_sys_name - " + audit_source_sys_name) if ("audit_src_sys_name".upper() in (name.upper() for name in dataframe.columns)): print("audit_src_sys_name column is present. Adding present layer details....") audit_dataframe = DataSink_with_audit(self.spark).add_audit_col_append(dataframe, audit_source_sys_name, write_params) print("Tables is updated.") else: print("Adding column - audit_src_sys_name.") audit_dataframe = DataSink_with_audit(self.spark).add_audit_col_fill(dataframe, audit_source_sys_name, write_params) print("audit_src_sys_name added.") print("Added Audit Cols") return audit_dataframe except Exception as e: raise Exception("job failed with error {}".format(e))

根据您提供的Python函数，我为您编写了一个MySQL存储过程的示例，但是需要根据您的具体需求进行修改和调整。这个存储过程的功能是向数据表中添加一个名为audit_src_sys_name的列，同时向这个列中填充相应的值。

CREATE PROCEDURE add_audit_columns(
    IN df_column_names VARCHAR(255),
    IN df_values VARCHAR(255),
    IN write_params TEXT,
    OUT audit_dataframe TEXT
)
BEGIN
    DECLARE audit_source_sys_name VARCHAR(255);
    DECLARE column_present BOOLEAN;
    DECLARE sql_stmt TEXT;
    
    SET audit_source_sys_name = JSON_EXTRACT(write_params, '$.audit_src_sys_name');
    IF audit_source_sys_name IS NULL OR audit_source_sys_name = '' THEN
        SET audit_source_sys_name = JSON_EXTRACT(write_params, '$.table_full_name');
    END IF;
    
    SET column_present = FALSE;
    SET sql_stmt = CONCAT('SELECT COUNT(*) INTO @exists FROM information_schema.columns WHERE table_name = \'', df_column_names, '\' AND column_name = \'audit_src_sys_name\'');
    PREPARE stmt FROM sql_stmt;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;
    
    IF @exists > 0 THEN
        SET column_present = TRUE;
        SET sql_stmt = CONCAT('UPDATE ', df_column_names, ' SET audit_src_sys_name = \'', audit_source_sys_name, '\'');
    ELSE
        SET sql_stmt = CONCAT('ALTER TABLE ', df_column_names, ' ADD COLUMN audit_src_sys_name VARCHAR(255) DEFAULT NULL AFTER ', df_values);
        PREPARE stmt FROM sql_stmt;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;
    END IF;
    
    IF column_present THEN
        SET audit_dataframe = CONCAT('Audit source system name column is present. Adding present layer details....\n');
        SET sql_stmt = CONCAT('UPDATE ', df_column_names, ' SET audit_src_sys_name = CONCAT(audit_src_sys_name, \', \', \'', audit_source_sys_name, '\')');
    ELSE
        SET audit_dataframe = CONCAT('Adding column - audit_src_sys_name.\n');
        SET sql_stmt = CONCAT('UPDATE ', df_column_names, ' SET audit_src_sys_name = \'', audit_source_sys_name, '\'');
    END IF;
    
    PREPARE stmt FROM sql_stmt;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;
    
    SET audit_dataframe = CONCAT(audit_dataframe, 'Tables is updated.\n');
END

这个存储过程的输入参数包括数据表的列名、值、以及一个JSON格式的字符串write_params，其中包含各种写入参数信息。输出参数是一个JSON格式的字符串audit_dataframe，包含存储过程的执行结果信息。请注意，这只是一个示例，您需要根据实际情况进行修改和调整。