micromegas_analytics/lakehouse/
processes_view.rs

1use super::{
2    session_configurator::NoOpSessionConfigurator, sql_batch_view::SqlBatchView,
3    view_factory::ViewFactory,
4};
5use anyhow::Result;
6use chrono::TimeDelta;
7use datafusion::execution::runtime_env::RuntimeEnv;
8use micromegas_ingestion::data_lake_connection::DataLakeConnection;
9use std::sync::Arc;
10
11/// Creates a new `SqlBatchView` for processes.
12pub async fn make_processes_view(
13    runtime: Arc<RuntimeEnv>,
14    lake: Arc<DataLakeConnection>,
15    view_factory: Arc<ViewFactory>,
16) -> Result<SqlBatchView> {
17    let count_src_query = Arc::new(String::from(
18        r#"
19        SELECT count(*) as count
20        FROM  blocks
21        WHERE insert_time >= '{begin}'
22        AND   insert_time < '{end}'
23        ;"#,
24    ));
25    let transform_query = Arc::new(String::from(
26        r#"
27SELECT process_id,
28       first_value("processes.exe") as exe,
29       first_value("processes.username") as username,
30       first_value("processes.realname") as realname,
31       first_value("processes.computer") as computer,
32       first_value("processes.distro") as distro,
33       first_value("processes.cpu_brand") as cpu_brand,
34       first_value("processes.tsc_frequency") as tsc_frequency,
35       first_value("processes.start_time") as start_time,
36       first_value("processes.start_ticks") as start_ticks,
37       first_value("processes.insert_time") as insert_time,
38       first_value("processes.parent_process_id") as parent_process_id,
39       first_value("processes.properties") as properties,
40       max(insert_time) as last_update_time,
41       max(end_ticks) as last_block_end_ticks,
42       max(end_time) as last_block_end_time,
43       arrow_cast(max(audience), 'Dictionary(Int32, Utf8)') as audience
44FROM blocks
45GROUP BY process_id
46        ;"#,
47    ));
48    let merge_query = Arc::new(String::from(
49        r#"
50SELECT process_id,
51       first_value("exe") as exe,
52       first_value("username") as username,
53       first_value("realname") as realname,
54       first_value("computer") as computer,
55       first_value("distro") as distro,
56       first_value("cpu_brand") as cpu_brand,
57       first_value("tsc_frequency") as tsc_frequency,
58       first_value("start_time") as start_time,
59       first_value("start_ticks") as start_ticks,
60       first_value("insert_time") as insert_time,
61       first_value("parent_process_id") as parent_process_id,
62       first_value("properties") as properties,
63       max(last_update_time) as last_update_time,
64       max(last_block_end_ticks) as last_block_end_ticks,
65       max(last_block_end_time) as last_block_end_time,
66       arrow_cast(max(audience), 'Dictionary(Int32, Utf8)') as audience
67FROM {source}
68GROUP BY process_id
69        ;"#,
70    ));
71    let min_time_column = Arc::new(String::from("insert_time"));
72    let max_time_column = Arc::new(String::from("last_update_time"));
73    SqlBatchView::new(
74        runtime,
75        Arc::new("processes".to_owned()),
76        min_time_column,
77        max_time_column,
78        count_src_query,
79        transform_query,
80        merge_query,
81        lake,
82        view_factory,
83        Arc::new(NoOpSessionConfigurator),
84        Some(2000),
85        TimeDelta::days(1), // from source
86        TimeDelta::days(1), // when merging
87    )
88    .await
89}
micromegas_analytics/lakehouse/processes_view.rs

micromegas_analytics/lakehouse/
processes_view.rs
