Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pandas/io/iceberg.py: 21%

1from typing import ( 

2 Any, 

3) 

4 

5from pandas.compat._optional import import_optional_dependency 

6from pandas.util._decorators import set_module 

7 

8from pandas import DataFrame 

9 

10 

11@set_module("pandas") 

12def read_iceberg( 

13 table_identifier: str, 

14 catalog_name: str | None = None, 

15 *, 

16 catalog_properties: dict[str, Any] | None = None, 

17 columns: list[str] | None = None, 

18 row_filter: str | None = None, 

19 case_sensitive: bool = True, 

20 snapshot_id: int | None = None, 

21 limit: int | None = None, 

22 scan_properties: dict[str, Any] | None = None, 

23) -> DataFrame: 

24 """ 

25 Read an Apache Iceberg table into a pandas DataFrame. 

26 

27 .. versionadded:: 3.0.0 

28 

29 .. warning:: 

30 

31 read_iceberg is experimental and may change without warning. 

32 

33 Parameters 

34 ---------- 

35 table_identifier : str 

36 Table identifier. 

37 catalog_name : str, optional 

38 The name of the catalog. 

39 catalog_properties : dict of {str: str}, optional 

40 The properties that are used next to the catalog configuration. 

41 columns : list of str, optional 

42 A list of strings representing the column names to return in the output 

43 dataframe. 

44 row_filter : str, optional 

45 A string that describes the desired rows. 

46 case_sensitive : bool, default True 

47 If True column matching is case sensitive. 

48 snapshot_id : int, optional 

49 Snapshot ID to time travel to. By default the table will be scanned as of the 

50 current snapshot ID. 

51 limit : int, optional 

52 An integer representing the number of rows to return in the scan result. 

53 By default all matching rows will be fetched. 

54 scan_properties : dict of {str: obj}, optional 

55 Additional Table properties as a dictionary of string key value pairs to use 

56 for this scan. 

57 

58 Returns 

59 ------- 

60 DataFrame 

61 DataFrame based on the Iceberg table. 

62 

63 See Also 

64 -------- 

65 read_parquet : Read a Parquet file. 

66 

67 Examples 

68 -------- 

69 >>> df = pd.read_iceberg( 

70 ... table_identifier="my_table", 

71 ... catalog_name="my_catalog", 

72 ... catalog_properties={"s3.secret-access-key": "my-secret"}, 

73 ... row_filter="trip_distance >= 10.0", 

74 ... columns=["VendorID", "tpep_pickup_datetime"], 

75 ... ) # doctest: +SKIP 

76 """ 

77 pyiceberg_catalog = import_optional_dependency("pyiceberg.catalog") 

78 pyiceberg_expressions = import_optional_dependency("pyiceberg.expressions") 

79 if catalog_properties is None: 

80 catalog_properties = {} 

81 catalog = pyiceberg_catalog.load_catalog(catalog_name, **catalog_properties) 

82 table = catalog.load_table(table_identifier) 

83 if row_filter is None: 

84 row_filter = pyiceberg_expressions.AlwaysTrue() 

85 if columns is None: 

86 selected_fields = ("*",) 

87 else: 

88 selected_fields = tuple(columns) # type: ignore[assignment] 

89 if scan_properties is None: 

90 scan_properties = {} 

91 result = table.scan( 

92 row_filter=row_filter, 

93 selected_fields=selected_fields, 

94 case_sensitive=case_sensitive, 

95 snapshot_id=snapshot_id, 

96 options=scan_properties, 

97 limit=limit, 

98 ) 

99 return result.to_pandas() 

100 

101 

102def to_iceberg( 

103 df: DataFrame, 

104 table_identifier: str, 

105 catalog_name: str | None = None, 

106 *, 

107 catalog_properties: dict[str, Any] | None = None, 

108 location: str | None = None, 

109 append: bool = False, 

110 snapshot_properties: dict[str, str] | None = None, 

111) -> None: 

112 """ 

113 Write a DataFrame to an Apache Iceberg table. 

114 

115 .. versionadded:: 3.0.0 

116 

117 Parameters 

118 ---------- 

119 table_identifier : str 

120 Table identifier. 

121 catalog_name : str, optional 

122 The name of the catalog. 

123 catalog_properties : dict of {str: str}, optional 

124 The properties that are used next to the catalog configuration. 

125 location : str, optional 

126 Location for the table. 

127 append : bool, default False 

128 If ``True``, append data to the table, instead of replacing the content. 

129 snapshot_properties : dict of {str: str}, optional 

130 Custom properties to be added to the snapshot summary 

131 

132 See Also 

133 -------- 

134 read_iceberg : Read an Apache Iceberg table. 

135 DataFrame.to_parquet : Write a DataFrame in Parquet format. 

136 """ 

137 pa = import_optional_dependency("pyarrow") 

138 pyiceberg_catalog = import_optional_dependency("pyiceberg.catalog") 

139 if catalog_properties is None: 

140 catalog_properties = {} 

141 catalog = pyiceberg_catalog.load_catalog(catalog_name, **catalog_properties) 

142 arrow_table = pa.Table.from_pandas(df) 

143 table = catalog.create_table_if_not_exists( 

144 identifier=table_identifier, 

145 schema=arrow_table.schema, 

146 location=location, 

147 # we could add `partition_spec`, `sort_order` and `properties` in the 

148 # future, but it may not be trivial without exposing PyIceberg objects 

149 ) 

150 if snapshot_properties is None: 

151 snapshot_properties = {} 

152 if append: 

153 table.append(arrow_table, snapshot_properties=snapshot_properties) 

154 else: 

155 table.overwrite(arrow_table, snapshot_properties=snapshot_properties)